// 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
}