johnandersen777.bsky.social/atlibs-socialweb-computer-requester
socialweb.computer market requester and policy modules for atlibs, with its web pages and lexicons
0
0
// One run for the whole contract: request -> bids -> policy -> accept -> receipt -> guest -> shell
// -> hold -> release -> cleanup. It reacts to each read as it lands and sets timers to re-read
// what is not there yet (the bid, the receipt, the guest's reports, the termination), so the
// caller does not poll. Records the run created are deleted at the end (cleanup, default on).
// release: false keeps the VM: the result carries `mcp` (servers for asyncai's mcp_connect) and
// `records`; a later `release {receipt}` (resume) runs the same tail: vm.delete, the bidder's
// vm.terminated, then deletes the kept records (`records`, else found from the receipt: its accept,
// the accept's rfp, the rfp's vm, those in the requester's repo), the release's own and the signals.
use std::collections::BTreeMap;
use abi::records::exec::{iroh, webrtc};
use abi::records::temp::compute::events::vm::Delete;
use abi::records::temp::market::evaluate_policy;
use abi::{AtUri, Call, Fan, Handle, On, Run, cap_token};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use crate::badge;
use crate::calls::{attest, b64url, read, typed};
use crate::calls::receipt_uri;
use crate::cleanup::{cleanup, signals};
use crate::guest::{GuestEvent, Reports, Stream, Via, report_refs, terminated};
use crate::ids::{CallId, Mode};
use crate::input::{Args, Polls, Route, Step, bidders, optional_route, peer, route, stdin, strong, strong_ref, tag, tags};
use crate::market::{AcceptChain, Receipt, RfpChain, denied, evaluate_policy, is_free, offers_compute, submit_rfp};
use crate::nsid::{BID, IROH, OFFERING, SUBMIT_RFP, WEBRTC};
use crate::release::PostChain;
#[derive(Clone, Copy, Default, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Phase {
#[default]
Request,
Keying,
Finding,
Accepting,
Accepted,
Guest,
Shell,
Holding,
Releasing,
Cleanup,
}
#[derive(Serialize, Deserialize)]
pub struct Offer {
did: String,
route: Route,
peer: Value,
}
#[derive(Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Winner {
bidder: String,
uri: String,
cid: String,
free: bool,
submit_accept: Value,
}
#[derive(Default, Serialize, Deserialize)]
pub struct Bid {
did: String,
name: String,
candidate: Option<Winner>,
value: Value,
}
#[derive(Default, Serialize, Deserialize)]
pub struct Release {
terminated: bool,
}
#[derive(Default, Serialize, Deserialize)]
#[serde(default, rename_all = "camelCase")]
pub struct Contract {
pub phase: Phase,
via: Option<Via>,
tag: String,
pub tags: Vec<String>,
command: Option<String>,
stdin: Option<String>,
bidders: Vec<String>,
offers: Vec<Option<Offer>>,
offerings: Fan,
pub request: RfpChain,
submitted: bool,
bids: BTreeMap<usize, Bid>,
bidders_in_play: usize,
winner: Option<Winner>,
accept: AcceptChain,
pub receipt: Option<Receipt>,
reports: Fan,
report_refs: Vec<AtUri>,
report_events: Vec<GuestEvent>,
guest: Option<Reports>,
shell: Option<Value>,
pub signals: Vec<Stream>,
release: Option<Release>,
post: PostChain,
pub created: Vec<String>,
pub deleted: Vec<String>,
pub deletes: Fan,
pub errors: Vec<String>,
polls: Polls,
key: String,
}
pub type Contracting = Step<Contract>;
pub fn contract(run: Run<Args>) -> Result<Handle, String> {
drive(run, start)
}
/// release {receipt, records?}: the tail of a contract kept with release: false.
pub fn resume(run: Run<Args>) -> Result<Handle, String> {
drive(run, kept)
}
fn drive(run: Run<Args>, start: fn(&mut Contracting) -> Result<(), String>) -> Result<Handle, String> {
let (mut step, on) = Contracting::new(run)?;
step.args.signed_in()?;
let c = &mut step;
match (c.state.phase, on) {
(_, On::Start) => keying(c),
(Phase::Keying, On::Result(id, result)) => keyed(c, id, result, start)?,
(Phase::Request, On::Result(CallId::Offerings(i), result)) => offering(c, i, result),
(Phase::Request, On::Result(id @ (CallId::UserData | CallId::Vm | CallId::AttestRfp | CallId::Rfp), result)) => rfp_created(c, id, result),
(Phase::Request, On::Result(CallId::SubmitRfp(i), result)) => rfp_submitted(c, i, result),
(Phase::Request, On::Result(CallId::Bid(i), result)) => bid(c, i, result),
(Phase::Request, On::Timer(CallId::Bid(i))) => read_bid(c, i),
(Phase::Request, On::Result(CallId::Scope(i), verdict)) => judged(c, Mode::Scope, i, verdict),
(Phase::Request, On::Result(CallId::Full(i), verdict)) => judged(c, Mode::Full, i, verdict),
(Phase::Accepting, On::Result(id, result)) => accepting(c, id, result),
(Phase::Accepted, On::Result(CallId::Receipt, result)) => receipt(c, result),
(Phase::Accepted, On::Timer(CallId::Receipt)) => read_receipt(c),
(Phase::Guest, On::Result(CallId::Report(k), result)) => report(c, k, result)?,
(Phase::Guest, On::Timer(CallId::Guest)) => read_guest(c),
(Phase::Shell, On::Result(CallId::Shell, result)) => shell(c, result),
(Phase::Cleanup, On::Result(CallId::Page(i), result)) => signals(c, i, result),
(Phase::Finding, On::Result(id, result)) => found(c, id, result),
(Phase::Holding, On::Timer(CallId::Hold)) => release(c),
(Phase::Releasing, On::Result(CallId::Term, result)) => term(c, result),
(Phase::Releasing, On::Timer(CallId::Term)) => read_term(c),
(Phase::Releasing, On::Result(id, result)) => releasing(c, id, result)?,
(Phase::Cleanup, On::Result(CallId::Delete(k), result)) => crate::cleanup::deleted(c, k, result),
_ => {}
}
step.handle()
}
// Before anything is signed: this host's attestation key must be bound to the account (badgeBlueKeys
// at the rkey verifiers read), or a bidder refuses what it signs (an RFP, an accept, a vm.delete). A
// key bound once is read, not written again; a host whose key is new (another runner, a page) binds it.
fn keying(c: &mut Contracting) {
c.state.phase = Phase::Keying;
c.out.call(CallId::Attest, attest(c.args.me(), None));
}
fn keyed(c: &mut Contracting, id: CallId, result: Result<Value, String>, start: fn(&mut Contracting) -> Result<(), String>) -> Result<(), String> {
let me = c.args.me().to_string();
match id {
CallId::Attest => {
let key = result?["key"].as_str().filter(|k| k.starts_with("did:key:")).ok_or("attest must return a did:key")?.to_string();
c.out.call(CallId::KeyBadge, read(&badge::binding(&me, &key)));
c.state.key = key;
return Ok(());
}
CallId::KeyBadge if !result.as_ref().is_ok_and(|o| o["value"].is_object()) => {
c.out.call(CallId::KeyBind, badge::bind(&me, &c.state.key, c.args.created_at()?));
return Ok(());
}
CallId::KeyBadge => {}
CallId::KeyBind => {
result.map_err(|e| format!("binding the attestation key {}: {e}", c.state.key))?;
}
_ => return Ok(()),
}
c.state.phase = Phase::Request;
start(c)
}
fn start(c: &mut Contracting) -> Result<(), String> {
let args = &c.args;
let bidders = bidders(args)?;
let via = match args.arg("via").as_str().unwrap_or("") {
"" => None,
"iroh" => Some(Via::Iroh),
"webrtc" => Some(Via::Webrtc),
_ => return Err("via must be iroh or webrtc".into()),
};
let tag = match args.arg("tag") {
Value::Null => "ssh".into(),
value => tag(value)?,
};
let command = match args.arg("command") {
Value::Null => None,
Value::String(c) if c.len() <= 4096 => Some(c.clone()),
_ => return Err("command must be a string of at most 4096 bytes".into()),
};
if command.is_some() && via.is_none() {
return Err("command needs via iroh or webrtc".into());
}
let stdin = stdin(args.arg("stdin"))?;
// The onNetwork reports read: the `tags` input (default ssh, shell), user_data's tunnels and `tag`.
let mut tags = tags(args.arg("tags"))?;
for t in tunnel_tags(args.arg("user_data").as_str().unwrap_or_default()).into_iter().chain([tag.clone()]) {
if !tags.contains(&t) {
tags.push(t);
}
}
let s = &mut c.state;
s.tags = tags;
c.out.fan_out(&mut s.offerings, bidders.iter().enumerate().map(|(i, bidder)| (CallId::Offerings(i), read(&AtUri::new(bidder, OFFERING, "self")))));
s.offers = bidders.iter().map(|_| None).collect();
(s.bidders, s.via, s.tag, s.command, s.stdin) = (bidders, via, tag, command, stdin);
s.request = RfpChain::start(args, None, &mut c.out)?;
Ok(())
}
// The kept contract's receipt (its result's {uri, cid, submitEvent}, or that JSON) and records.
fn kept(c: &mut Contracting) -> Result<(), String> {
let receipt = match c.args.arg("receipt") {
Value::String(text) => serde_json::from_str(text).map_err(|_| "receipt must be a strong ref {uri, cid}")?,
value => value.clone(),
};
let r = strong(&receipt, "receipt")?;
let bidder = r.uri.parse::<AtUri>().map(|u| u.repo).map_err(|_| "receipt must be a strong ref {uri, cid}")?;
let mine = format!("at://{}/", c.args.me());
let records = match c.args.arg("records") {
Value::Null => None,
Value::Array(list) if list.len() <= 64 => Some(list.iter().map(|v| v.as_str().filter(|u| u.starts_with(&mine)).map(str::to_string)).collect::<Option<Vec<_>>>()),
_ => Some(None),
};
let s = &mut c.state;
let submit_event = Some(c.args.arg("submitEvent").clone()).filter(|v| !v.is_null()).unwrap_or(receipt["submitEvent"].clone());
s.receipt = Some(Receipt { uri: r.uri.clone(), cid: json!(r.cid), submit_event });
s.winner = Some(Winner { bidder, uri: String::new(), cid: String::new(), free: false, submit_accept: Value::Null });
match records {
Some(records) => {
s.created = records.ok_or("records must be a list of at most 64 at:// URIs in your repo")?;
// this is the release: a composition's `release: false` (for its contract) does not stop it
give_back(c);
}
None => {
s.phase = Phase::Finding;
c.out.call(CallId::Receipt, read(&r.uri.parse::<AtUri>().map_err(|e| e.to_string())?));
}
}
Ok(())
}
// Finding (release without records): the receipt names the accept, the accept the rfp, the rfp
// its vm. Each in the requester's repo is kept (deleted after the release); a read that fails or
// a ref outside the repo ends the search, noted, and the release goes ahead with what was found.
fn found(c: &mut Contracting, id: CallId, result: Result<Value, String>) {
let (key, next) = match id {
CallId::Receipt => ("accept", Some(CallId::Accept)),
CallId::Accept => ("rfp", Some(CallId::Rfp)),
CallId::Rfp => ("payload", None),
_ => return,
};
let output = match result {
Ok(output) if output["value"].is_object() => output,
Ok(_) => return found_end(c, format!("finding the records: {id} has no value")),
Err(e) => return found_end(c, format!("finding the records: {id}: {e}")),
};
let s = &mut c.state;
if let (CallId::Receipt, Some(receipt)) = (id, s.receipt.as_mut()) {
if receipt.submit_event.is_null() {
receipt.submit_event = output["value"]["submitEvent"].clone();
}
}
let mine = format!("at://{}/", c.args.me());
let uri = output["value"][key]["uri"].as_str().unwrap_or_default();
let Some(uri) = uri.parse::<AtUri>().ok().filter(|_| uri.starts_with(&mine)) else {
return found_end(c, format!("finding the records: {id}'s {key} {} is not in your repo", json!(uri)));
};
s.created.insert(0, uri.to_string());
match next {
Some(next) => c.out.call(next, read(&uri)),
None => give_back(c),
}
}
fn found_end(c: &mut Contracting, error: String) {
note(c, error);
give_back(c);
}
fn note(c: &mut Contracting, error: String) {
c.state.errors.push(error);
}
// ---- request: offerings and the rfp, then submitRfp to every bidder with a compute offering -------
fn offering(c: &mut Contracting, i: usize, result: Result<Value, String>) {
let bidder = c.state.bidders.get(i).cloned().unwrap_or_default();
match &result {
Ok(output) => match Some(&output["value"]).filter(|v| offers_compute(v)).and_then(|v| peer(&bidder, v)) {
Some(peer) => match route(&peer, "endpoints") {
Ok(route) => c.state.offers[i] = Some(Offer { did: bidder, route, peer }),
Err(e) => note(c, format!("{bidder}: {e}")),
},
None => note(c, format!("{bidder}: no compute offering")),
},
Err(e) => note(c, format!("{bidder} offerings: {e}")),
}
if c.state.offerings.land(&CallId::Offerings(i), result).is_some() {
submit(c);
}
}
// Any failure in the request, accept and release chains goes through fail, so cleanup runs.
fn rfp_created(c: &mut Contracting, id: CallId, result: Result<Value, String>) {
match c.state.request.step(&c.args, id, result, &mut c.out, &mut c.state.created) {
Ok(Some(_)) => submit(c),
// the hook's user_data: its tunnels' reports are read too
Ok(None) if id == CallId::UserData => {
let s = &mut c.state;
for t in tunnel_tags(&s.request.user_data) {
if !s.tags.contains(&t) {
s.tags.push(t);
}
}
}
Ok(None) => {}
Err(e) => fail(c, e),
}
}
fn submit(c: &mut Contracting) {
let s = &mut c.state;
let Some(rfp) = &s.request.rfp else { return };
if s.submitted || !s.offerings.is_empty() {
return;
}
s.submitted = true;
let input = submit_rfp(rfp);
let submits: Vec<_> = s.offers.iter().enumerate().filter_map(|(i, offer)| Some((CallId::SubmitRfp(i), offer.as_ref()?.route.call(SUBMIT_RFP, &input)))).collect();
s.bidders_in_play = submits.len();
if submits.is_empty() {
return fail(c, "no bidder offers compute".into());
}
for (id, call) in submits {
c.out.call(id, call);
}
}
fn rfp_submitted(c: &mut Contracting, i: usize, result: Result<Value, String>) {
let did = c.state.offers.get(i).and_then(Option::as_ref).map(|o| o.did.clone()).unwrap_or_default();
match result.as_ref().map(|output| output["name"].as_str()) {
Ok(Some(name)) => {
c.state.bids.insert(i, Bid { did, name: name.into(), ..Bid::default() });
read_bid(c, i);
}
failed => {
let why = failed.err().cloned().unwrap_or("no bid name".into());
note(c, format!("{did} submitRfp: {why}"));
no_bid_left(c);
}
}
}
// ---- bids: read each until it shows up; the policy hook judges the bidder, then the bid ----------
fn read_bid(c: &mut Contracting, i: usize) {
if let Some(bid) = c.state.bids.get(&i) {
c.out.call(CallId::Bid(i), read(&AtUri::new(&bid.did, BID, &bid.name)));
}
}
fn bid(c: &mut Contracting, i: usize, result: Result<Value, String>) {
let s = &mut c.state;
let (Some(rfp), Some(bid)) = (&s.request.rfp, s.bids.get_mut(&i)) else { return };
match result {
Ok(output) if output["value"]["rfp"]["uri"] == rfp.uri.as_str() => {
let value = &output["value"];
let free = is_free(&value["payload"]);
if !free && c.args.arg("accept_paid") != true {
note(c, format!("bid {} is not free", output["uri"]));
return no_bid_left(c);
}
let text = |v: &Value| v.as_str().unwrap_or_default().to_string();
let winner = Winner { bidder: bid.did.clone(), uri: text(&output["uri"]), cid: text(&output["cid"]), free, submit_accept: s.offers.get(i).and_then(Option::as_ref).map_or(Value::Null, |o| o.peer.clone()) };
c.out.call(CallId::Scope(i), evaluate_policy(c.args.me(), Mode::Scope, rfp, &winner.bidder, None));
(bid.candidate, bid.value) = (Some(winner), value.clone());
}
_ if s.polls.again(&c.args, &mut c.out, CallId::Bid(i)) => {}
_ => no_bid_left(c),
}
}
fn judged(c: &mut Contracting, mode: Mode, i: usize, verdict: Result<Value, String>) {
let s = &mut c.state;
let (Some(rfp), Some(bid)) = (&s.request.rfp, s.bids.get(&i)) else { return };
let Some(candidate) = bid.candidate.clone() else { return };
if let Some(why) = denied(&verdict) {
note(c, format!("bid {} denied by policy ({mode}): {why}", candidate.uri));
return no_bid_left(c);
}
match mode {
Mode::Scope => {
let record = evaluate_policy::Record { uri: candidate.uri.clone(), cid: Some(candidate.cid.clone()), value: Some(bid.value.clone()).filter(|v| !v.is_null()) };
c.out.call(CallId::Full(i), evaluate_policy(c.args.me(), Mode::Full, rfp, &candidate.bidder, Some(record)));
}
Mode::Full => accept(c, candidate),
}
}
// A bidder dropped out (submitRfp failed, its bid never showed up, was not free, or was denied).
fn no_bid_left(c: &mut Contracting) {
c.state.bidders_in_play = c.state.bidders_in_play.saturating_sub(1);
if c.state.bidders_in_play == 0 && c.state.winner.is_none() {
fail(c, "no acceptable bid".into());
}
}
// ---- accept, then the receipt the winner mints for it -------------------------------------------
fn accept(c: &mut Contracting, winner: Winner) {
let s = &mut c.state;
let Some(rfp) = s.request.rfp.clone() else { return };
let bid = match strong_ref(Some(&winner.uri), Some(&winner.cid), "bid") {
Ok(bid) => bid,
Err(e) => return fail(c, e),
};
let submit_accept = optional_route(&winner.submit_accept, "submitAccept").ok().flatten();
s.winner = Some(winner);
s.phase = Phase::Accepting;
match AcceptChain::start(&c.args, rfp, bid, submit_accept, &mut c.out) {
Ok(chain) => s.accept = chain,
Err(e) => fail(c, e),
}
}
fn accepting(c: &mut Contracting, id: CallId, result: Result<Value, String>) {
match c.state.accept.step(&c.args, id, result, &mut c.out, &mut c.state.created) {
Ok(Some(accepted)) => {
if let Some(e) = accepted.error {
note(c, format!("submitAccept: {e}"));
}
c.state.phase = Phase::Accepted;
read_receipt(c);
}
Ok(None) => {}
Err(e) => fail(c, e),
}
}
fn read_receipt(c: &mut Contracting) {
let s = &c.state;
let (Some(winner), Some(accept)) = (&s.winner, &s.accept.accept) else { return };
c.out.call(CallId::Receipt, read(&receipt_uri(&winner.bidder, accept)));
}
fn receipt(c: &mut Contracting, result: Result<Value, String>) {
let s = &mut c.state;
match result {
Ok(output) if output["value"].is_object() => {
s.receipt = Some(Receipt::of(&output));
s.phase = Phase::Guest;
progress(c);
read_guest(c);
}
_ if s.polls.again(&c.args, &mut c.out, CallId::Receipt) => {}
_ => fail(c, "the winner minted no receipt".into()),
}
}
// progress (a callback the caller passed): the round's refs as soon as a VM exists, so a
// caller that outlives this run can still release it. Not waited for.
fn progress(c: &mut Contracting) {
let Some(token) = cap_token(c.args.arg("progress")).filter(|t| t.starts_with("cap.")) else { return };
let s = &c.state;
let refs = json!({ "vm": s.request.vm, "rfp": s.request.rfp, "accept": s.accept.accept, "receipt": s.receipt });
c.out.call(CallId::Progress, Call::new("", token).input(refs));
}
// ---- guest: its reports until what `via` needs is there, then the command -----------------------
fn read_guest(c: &mut Contracting) {
let s = &mut c.state;
let (Some(winner), Some(receipt)) = (&s.winner, &s.receipt) else { return };
s.report_refs = report_refs(&winner.bidder, &receipt.uri, &s.tags);
s.report_events.clear();
c.out.fan_out(&mut s.reports, s.report_refs.iter().enumerate().map(|(k, uri)| (CallId::Report(k), read(uri))));
}
fn report(c: &mut Contracting, k: usize, result: Result<Value, String>) -> Result<(), String> {
let s = &mut c.state;
s.report_events.extend(s.report_refs.get(k).and_then(|uri| GuestEvent::found(&uri.to_string(), &result)));
if s.reports.land(&CallId::Report(k), result).is_none() {
return Ok(());
}
let reports = Reports::of(&s.report_events);
let ready = reports.ready(s.via);
s.guest = Some(reports);
if ready {
return run_command(c);
}
if !s.polls.again(&c.args, &mut c.out, CallId::Guest) {
let via = match s.via {
Some(Via::Iroh) => "iroh",
Some(Via::Webrtc) => "webrtc",
None => "",
};
// never came up: not a VM to keep, whatever `release` says; give it back and delete the records
note(c, format!("the guest did not report what via {via:?} needs"));
give_back(c);
}
Ok(())
}
fn run_command(c: &mut Contracting) -> Result<(), String> {
let s = &mut c.state;
let Some(command) = s.command.clone() else {
hold(c);
return Ok(());
};
let call = match s.via {
Some(Via::Iroh) => {
let guest = s.guest.clone().unwrap_or_default();
let input = iroh::Connect { ticket: guest.iroh.unwrap_or_default(), alpn: None, host_keys: guest.host_keys, command: Some(command), stdin: s.stdin.clone(), send: None };
Call::new("", IROH).input(json!(input))
}
_ => {
let receipt = s.receipt.as_ref();
let receipt = strong_ref(receipt.map(|r| r.uri.as_str()), receipt.and_then(|r| r.cid.as_str()), "receipt")?;
let input = webrtc::Connect { receipt, tag: s.tag.clone(), command: Some(command), stdin: s.stdin.clone() };
Call::new("", WEBRTC).input(json!(input))
}
};
s.phase = Phase::Shell;
c.out.call(CallId::Shell, call);
Ok(())
}
fn shell(c: &mut Contracting, result: Result<Value, String>) {
let shell = match result {
Ok(output) => output,
Err(e) => {
note(c, format!("the shell failed: {e}"));
json!({ "error": e })
}
};
c.state.shell = Some(shell);
hold(c);
}
pub fn hold(c: &mut Contracting) {
let ms = c.args.number("hold_ms", 0, 3_600_000);
if ms == 0 {
return release(c);
}
c.out.timer(CallId::Hold, ms);
c.state.phase = Phase::Holding;
}
// ---- release: vm.delete, then the bidder's vm.terminated --------------------------------------
// release: false keeps the VM (the result carries the receipt for a later release).
fn release(c: &mut Contracting) {
if *c.args.arg("release") == false {
return finish(c);
}
give_back(c);
}
// vm.delete, then the bidder's vm.terminated (bounded by max_polls), then cleanup.
fn give_back(c: &mut Contracting) {
let s = &mut c.state;
s.phase = Phase::Releasing;
s.release = Some(Release { terminated: false });
let receipt = s.receipt.as_ref();
// the destroy goes to the winner (its offering endpoints), else to the receipt's submitEvent, else only as a record
let winner = s.winner.as_ref().and_then(|w| optional_route(&w.submit_accept, "submitAccept").ok().flatten());
let submit_event = winner.or_else(|| receipt.and_then(|r| optional_route(&r.submit_event, "submitEvent").ok().flatten()));
let reason = c.args.arg("reason").as_str().filter(|r| !r.is_empty()).unwrap_or("session_ended");
let posting = strong_ref(receipt.map(|r| r.uri.as_str()), receipt.and_then(|r| r.cid.as_str()), "receipt").and_then(|receipt| {
let payload = typed(Delete { reason: reason.into(), created_at: c.args.created_at()? });
Ok(PostChain::start(&c.args, receipt, submit_event, payload, None, &mut c.out))
});
match posting {
Ok(post) => s.post = post,
Err(e) => fail_release(c, e),
}
}
// A bidder that refuses the vm.delete (submitEvent fails) still runs the VM: the release fails, and
// its records (the vm.delete and event among them) are kept for another try.
fn releasing(c: &mut Contracting, id: CallId, result: Result<Value, String>) -> Result<(), String> {
match c.state.post.step(&c.args, id, result, &mut c.out, &mut c.state.created) {
Ok(Some(posted)) => match posted.error {
Some(e) => {
let kept: Vec<&String> = c.state.created.iter().filter(|u| !c.state.deleted.contains(u)).collect();
return Err(format!("submitEvent: {e}; the VM is not released, these records are kept: {}", json!(kept)));
}
None => read_term(c),
},
Ok(None) => {}
Err(e) => fail_release(c, e),
}
Ok(())
}
fn read_term(c: &mut Contracting) {
let (Some(winner), Some(receipt)) = (&c.state.winner, &c.state.receipt) else { return };
c.out.call(CallId::Term, read(&terminated(&winner.bidder, &receipt.uri)));
}
fn term(c: &mut Contracting, result: Result<Value, String>) {
let s = &mut c.state;
match result {
Ok(output) if output["value"].is_object() => {
s.release.get_or_insert_default().terminated = true;
cleanup(c);
}
_ if s.polls.again(&c.args, &mut c.out, CallId::Term) => {}
_ => {
note(c, "the bidder reported no vm.terminated".into());
cleanup(c);
}
}
}
// ---- endings ------------------------------------------------------------------------------------
// Without a receipt nothing runs on the bidder (no accept yet, or the winner never minted one), and
// release needs the receipt: clean up and end. With one, give it back. Either way, whatever
// `release` says: a failed rent is not a VM to keep.
fn fail(c: &mut Contracting, error: String) {
note(c, error);
if c.state.receipt.is_none() {
return cleanup(c);
}
give_back(c);
}
fn fail_release(c: &mut Contracting, error: String) {
note(c, format!("release: {error}"));
cleanup(c);
}
pub fn finish(c: &mut Contracting) {
let s = &c.state;
let release = s.release.as_ref().map(|r| json!({ "terminated": r.terminated, "payload": s.post.payload, "event": s.post.event }));
let records: Vec<&String> = s.created.iter().filter(|u| !s.deleted.contains(u)).collect();
let mcp = if s.release.is_none() { mcp_servers(s) } else { vec![] };
let result = json!({
"vm": s.request.vm, "rfp": s.request.rfp, "bid": s.winner, "accept": s.accept.accept, "receipt": s.receipt,
"guest": s.guest, "shell": s.shell, "release": release, "deleted": s.deleted, "records": records, "mcp": mcp, "errors": s.errors,
});
c.out.finish(result);
}
pub const SHELL_TAG: &str = "shell";
pub const TUNNELS_PATH: &str = "/etc/socialweb-guest-agent/tunnels.json";
// The tags of the tunnels.json a JSON #cloud-config writes (the user-data module); other user_data: none.
fn tunnel_tags(user_data: &str) -> Vec<String> {
let Some(doc) = user_data.strip_prefix("#cloud-config\n").and_then(|j| serde_json::from_str::<Value>(j).ok()) else { return vec![] };
let files = doc["write_files"].as_array().into_iter().flatten().filter(|f| f["path"] == TUNNELS_PATH);
let tunnels = files.filter_map(|f| serde_json::from_str::<Value>(f["content"].as_str()?).ok());
tunnels.flat_map(|t| t["tunnels"].as_array().cloned().unwrap_or_default()).filter_map(|t| tag(&t["tag"]).ok()).collect()
}
// The kept VM's guest-agent shell MCP server, per transport the guest reported: webrtc (the shell
// tag's fingerprint) and iroh (a ticket and pinned host keys). Each is asyncai's mcp_connect
// {name, server: {url}}: webrtc:///<b64url({receipt, bidder})>/shell/mcp and
// iroh:///<b64url({ticket, hostKeys})>/shell/mcp, fetched by the host's scheme fetchers.
fn mcp_servers(s: &Contract) -> Vec<Value> {
let (Some(receipt), Some(guest)) = (&s.receipt, &s.guest) else { return vec![] };
let rkey = receipt.uri.rsplit('/').next().unwrap_or_default();
let name = format!("vm-{}", rkey.strip_prefix("receipt-").unwrap_or(rkey).chars().take(8).collect::<String>());
let url = |scheme: &str, r: Value| format!("{scheme}:///{}/{SHELL_TAG}/mcp", b64url(r.to_string().as_bytes()));
let mut servers = vec![];
if guest.webrtc.iter().any(|w| w.tag == SHELL_TAG) {
let bidder = s.winner.as_ref().map(|w| w.bidder.clone());
let r = json!({ "receipt": { "uri": receipt.uri, "cid": receipt.cid }, "bidder": bidder });
servers.push(json!({ "name": name, "server": { "url": url("webrtc", r) } }));
}
if let (Some(ticket), true) = (&guest.iroh, guest.pinned()) {
let r = json!({ "ticket": ticket, "hostKeys": guest.host_keys });
servers.push(json!({ "name": format!("{name}-iroh"), "server": { "url": url("iroh", r) } }));
}
servers
}