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