use abi::repo::{delete_record, get_record, put_record};
use abi::{AtUri, Call, Event, Handle, Run, entries};
use serde::Deserialize;
use serde_json::{Value, json};

// An atirc agent on a VM you keep, on the atirc mesh (no atirc server, no token): two mesh nodes on the
// VM, the agent (<nick>.<handle>, `atirc agent run` beside it) and yours there (you-<nick>.<handle>), both
// in your channel and the agent's own. You write their computer.socialweb.peer records (iroh: the bare
// endpoint id of a key the VM made); the agents you run are one computer.socialweb.agents/self record.
const PEER: &str = "computer.socialweb.peer";
const ASSOCIATION: &str = "com.publicdomainrelay.temp.market.association";
const INDEX: &str = "computer.socialweb.agents";
const EXEC: &str = "computer.socialweb.cicd.process.exec";
const PREPARE: &str = include_str!("../prepare.sh");
const START: &str = include_str!("../start.sh");
const STOP: &str = "set -eu; read -r n; read -r y; systemctl disable --now atirc-agent-$n atirc-mesh-$n atirc-mesh-$y 2>/dev/null || true; rm -f /etc/systemd/system/atirc-agent-$n.service /etc/systemd/system/atirc-mesh-$n.service /etc/systemd/system/atirc-mesh-$y.service; systemctl daemon-reload; echo stopped";
const HARNESS: &str = r#"#!/usr/bin/env bash
export ANTHROPIC_MODEL="$(atirc agent anthropic model)"
export ANTHROPIC_BASE_URL="$(atirc agent anthropic base-url)"
export ANTHROPIC_API_KEY="$(atirc agent anthropic api-key)"
export ANTHROPIC_DEFAULT_OPUS_MODEL="${ANTHROPIC_MODEL}"
export ANTHROPIC_DEFAULT_SONNET_MODEL="${ANTHROPIC_MODEL}"
export ANTHROPIC_DEFAULT_HAIKU_MODEL="${ANTHROPIC_MODEL}"
export CLAUDE_CODE_SUBAGENT_MODEL="${ANTHROPIC_MODEL}"
command -v claude >/dev/null 2>&1 || curl -fsSL https://claude.ai/install.sh | bash
exec claude --allow-dangerously-skip-permissions --dangerously-skip-permissions "$@""#;
const MAX_AGENTS: usize = 20;
// your node's local listener on the VM (atirc mesh run --local-listen), reached over the contract's "atirc" tunnel
const LISTEN_PORT: u64 = 6680;

#[derive(Deserialize, Default)]
struct Args {
    #[serde(default)]
    did: String,
    #[serde(default)]
    now: String,
    #[serde(default)]
    of: Option<String>,
    #[serde(default)]
    vm: Option<String>,
    #[serde(default)]
    shell: Option<String>,
    #[serde(default)]
    nick: Option<String>,
    #[serde(default)]
    model: Option<String>,
    #[serde(default)]
    base_url: Option<String>,
    #[serde(default)]
    expires: Option<String>,
    #[serde(default)]
    listen_port: Option<u64>,
    #[serde(default)]
    api_key: Option<String>,
    #[serde(default)]
    harness: Option<String>,
    #[serde(default)]
    atirc_url: Option<String>,
    #[serde(default)]
    atirc_sha256: Option<String>,
}

/// atirc's channel for a handle or subhandle: '#' + it, '.' -> '_', lowercase.
pub fn channel(name: &str) -> String {
    let c: String = name.chars().take(63).map(|c| if c == '.' || c == ',' || c.is_whitespace() { '_' } else { c }).collect();
    format!("#{}", c.to_ascii_lowercase())
}

fn nick_ok(n: &str) -> bool {
    let b = n.as_bytes();
    !b.is_empty() && b.len() <= 23 && b[0].is_ascii_lowercase() && b.iter().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || *c == b'-') && !n.starts_with("you-")
}

// A value that goes into config.toml inside "...": no quote, backslash or control character.
fn plain(key: &str, v: &Option<String>) -> Result<String, String> {
    let v = v.as_deref().map(str::trim).unwrap_or_default();
    if v.is_empty() || v.len() > 2048 || v.chars().any(|c| c == '"' || c == '\\' || c.is_control()) {
        return Err(format!("{key}: required, no quotes, backslashes or control characters"));
    }
    Ok(v.to_string())
}

fn b64(s: &str) -> String {
    const A: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    let mut out = String::new();
    for chunk in s.as_bytes().chunks(3) {
        let n = chunk.iter().enumerate().fold(0u32, |n, (i, b)| n | (*b as u32) << (16 - 8 * i));
        for i in 0..4 {
            out.push(if i <= chunk.len() { A[(n >> (18 - 6 * i)) as usize & 63] as char } else { '=' });
        }
    }
    out
}

fn b64url_decode(s: &str) -> Option<Vec<u8>> {
    let mut bits = 0u32;
    let mut n = 0;
    let mut out = Vec::new();
    for c in s.trim_end_matches('=').bytes() {
        let v = match c {
            b'A'..=b'Z' => c - b'A',
            b'a'..=b'z' => c - b'a' + 26,
            b'0'..=b'9' => c - b'0' + 52,
            b'-' | b'+' => 62,
            b'_' | b'/' => 63,
            _ => return None,
        } as u32;
        bits = bits << 6 | v;
        n += 6;
        if n >= 8 {
            n -= 8;
            out.push((bits >> n) as u8);
        }
    }
    Some(out)
}

/// The receipt a webrtc:/// or iroh:/// shell URL names: its ref is base64url JSON {receipt: {uri, cid}, ...}.
fn receipt_of(shell: &str) -> Option<String> {
    let reference = shell.split('/').nth(3)?;
    let v: Value = serde_json::from_slice(&b64url_decode(reference)?).ok()?;
    v["receipt"]["uri"].as_str().filter(|u| u.contains("/com.publicdomainrelay.temp.market.receipt/")).map(str::to_string)
}

/// kcp's contract-scoped association key: requester_associate-<first 32 hex of sha256(subject)>-<accept rkey>.
fn lease_rkey(subject: &str, accept_uri: &str) -> String {
    use sha2::Digest;
    let hex: String = sha2::Sha256::digest(subject.as_bytes()).iter().take(16).map(|b| format!("{b:02x}")).collect();
    format!("requester_associate-{hex}-{}", accept_uri.rsplit('/').next().unwrap_or_default())
}

fn prepare_stdin(a: &Args, s: &Value) -> String {
    format!(
        "NICK={}\nYOU={}\nATIRC_URL={}\nATIRC_SHA256={}\nMODEL_B64={}\nBASE_URL_B64={}\nAPI_KEY_B64={}\nHARNESS_B64={}\n",
        s["nick"].as_str().unwrap_or_default(),
        s["you"].as_str().unwrap_or_default(),
        s["atirc_url"].as_str().unwrap_or_default(),
        s["atirc_sha256"].as_str().unwrap_or_default(),
        b64(a.model.as_deref().unwrap_or_default().trim()),
        b64(a.base_url.as_deref().unwrap_or_default().trim()),
        b64(a.api_key.as_deref().unwrap_or_default().trim()),
        b64(s["harness"].as_str().unwrap_or_default()),
    )
}

fn exec(id: &str, url: &str, command: &str, stdin: String, timeout_ms: u64) -> Call {
    Call::new(id, EXEC).input(json!({ "url": url, "command": command, "stdin": stdin, "timeoutMs": timeout_ms }))
}

fn read(id: &str, repo: &str, collection: &str, rkey: &str) -> Call {
    get_record(&AtUri::new(repo, collection, rkey)).id(id).service(repo)
}

fn agents_of(result: &abi::CallResult) -> Vec<Value> {
    result.output().ok().and_then(|o| o["value"]["agents"].as_array().cloned()).unwrap_or_default()
}

fn exec_out(result: &abi::CallResult, what: &str) -> Result<String, String> {
    let out = result.output().map_err(|e| format!("{what}: {e}"))?;
    if out["exitCode"].as_i64() != Some(0) {
        return Err(format!("{what} exited {}: {}", out["exitCode"], out["stderr"].as_str().unwrap_or_default().chars().rev().take(600).collect::<String>().chars().rev().collect::<String>()));
    }
    Ok(out["stdout"].as_str().unwrap_or_default().to_string())
}

fn index_put(did: &str, agents: Vec<Value>, now: &str) -> Call {
    put_record(&AtUri::new(did, INDEX, "self"), json!({ "$type": INDEX, "agents": agents, "updatedAt": now })).id("index-put")
}

// launch {vm, shell, nick, model, base_url, api_key, harness?}: returns {nick, subhandle, you, channels, vm}.
fn launch_handler(run: Run<Args>) -> Result<Handle, String> {
    let a = &run.args;
    let me = a.did.clone();
    if !me.starts_with("did:") {
        return Err("sign in first".into());
    }
    let mut s = run.state.clone();
    Ok(match run.event {
        Event::Start => {
            let nick = a.nick.clone().unwrap_or_default().trim().to_ascii_lowercase();
            if !nick_ok(&nick) {
                return Err("nick: a lowercase letter, then letters, digits or -, at most 23, not you-*".into());
            }
            let shell = a.shell.clone().filter(|u| u.starts_with("webrtc:///") || u.starts_with("iroh:///")).ok_or("shell: the VM's shell URL (vm_list)")?;
            let base = plain("base_url", &a.base_url)?;
            if !base.starts_with("https://") && !base.starts_with("http://") {
                return Err("base_url: an http(s) URL".into());
            }
            plain("model", &a.model)?;
            plain("api_key", &a.api_key)?;
            let harness = a.harness.clone().filter(|h| !h.trim().is_empty()).unwrap_or(HARNESS.into());
            if !harness.starts_with("#!") || harness.contains("'''") {
                return Err("harness: a #! script without '''".into());
            }
            let sha = a.atirc_sha256.clone().filter(|h| h.len() == 64 && h.bytes().all(|b| b.is_ascii_hexdigit())).ok_or("atirc_sha256: the pinned binary's sha256")?;
            let url = a.atirc_url.clone().filter(|u| u.starts_with("https://") && !u.contains(char::is_whitespace)).ok_or("atirc_url: an https URL")?;
            let receipt = receipt_of(&shell).ok_or("shell: its ref names no market receipt")?;
            let expires = a.expires.clone().filter(|e| e.len() >= 20 && e.as_bytes()[4] == b'-' && e.contains('T')).ok_or("expires: when the VM ends (vm_list's expires)")?;
            let listen_port = a.listen_port.unwrap_or(LISTEN_PORT);
            if !(1024..=65535).contains(&listen_port) {
                return Err("listen_port: 1024-65535".into());
            }
            let state = json!({ "listen_port": listen_port, "nick": nick, "you": format!("you-{nick}"), "shell": shell, "receipt": receipt, "expires": expires, "vm": a.vm, "harness": harness, "atirc_url": url, "atirc_sha256": sha });
            Handle::new().call(Call::new("repo", "com.atproto.repo.describeRepo").params(json!({ "repo": me })).service(&me)).state(state)
        }
        Event::Result { id, result, .. } => match id.as_str() {
            "repo" => {
                let handle = result.output()?["handle"].as_str().filter(|h| h.contains('.')).ok_or("your handle did not resolve")?.to_string();
                s["handle"] = json!(handle);
                Handle::new().call(read("index", &me, INDEX, "self")).state(s)
            }
            "index" => {
                let agents = agents_of(&result);
                let nick = s["nick"].as_str().unwrap_or_default().to_string();
                if agents.iter().any(|x| x["nick"] == nick.as_str()) {
                    return Err(format!("{nick} is running already: stop it first"));
                }
                if agents.len() >= MAX_AGENTS {
                    return Err(format!("at most {MAX_AGENTS} agents"));
                }
                s["agents"] = json!(agents);
                let receipt = s["receipt"].as_str().unwrap_or_default().to_string();
                let bidder = receipt.split('/').nth(2).unwrap_or_default().to_string();
                Handle::new().call(get_record(&receipt.parse::<AtUri>().map_err(|_| "bad receipt uri")?).id("receipt").service(&bidder)).state(s)
            }
            // the lease (kcp 0004): a contract-scoped association of your own DID, ending with the VM
            "receipt" => {
                let accept = &result.output()?["value"]["accept"];
                let (uri, cid) = (accept["uri"].as_str().unwrap_or_default(), accept["cid"].as_str().unwrap_or_default());
                if !uri.starts_with(&format!("at://{me}/com.publicdomainrelay.temp.market.accept/")) || cid.is_empty() {
                    return Err("the VM's receipt names no accept of yours".into());
                }
                let rkey = lease_rkey(&me, uri);
                let record = json!({ "$type": ASSOCIATION, "subject": me, "service": "requester_associate", "contract": { "uri": uri, "cid": cid }, "expiresAt": s["expires"], "note": format!("atirc agent {}", s["nick"].as_str().unwrap_or_default()), "createdAt": run.args.now });
                Handle::new().call(put_record(&AtUri::new(&me, ASSOCIATION, &rkey), record).id("lease")).state(s)
            }
            "lease" => {
                let out = result.output()?;
                s["lease"] = json!({ "uri": out["uri"], "cid": out["cid"] });
                let shell = s["shell"].as_str().unwrap_or_default().to_string();
                Handle::new().call(exec("prepare", &shell, PREPARE, prepare_stdin(a, &s), 300_000)).state(s)
            }
            "prepare" => {
                let out = exec_out(&result, "prepare")?;
                let ids: Value = serde_json::from_str(out.trim().lines().last().unwrap_or_default()).map_err(|_| format!("prepare printed no ids: {out}"))?;
                let handle = s["handle"].as_str().unwrap_or_default().to_string();
                let nick = s["nick"].as_str().unwrap_or_default().to_string();
                let you = s["you"].as_str().unwrap_or_default().to_string();
                let channels = json!([channel(&handle), channel(&format!("{nick}.{handle}"))]);
                for (k, id) in [("agent", &ids["agent"]), ("you", &ids["you"])] {
                    let id = id.as_str().filter(|i| i.len() == 64 && i.bytes().all(|b| b.is_ascii_hexdigit())).ok_or(format!("prepare: no {k} endpoint id"))?;
                    s[format!("{k}_id")] = json!(id);
                }
                s["channels"] = channels.clone();
                let lease = s["lease"].clone();
                let record = |id: &Value| json!({ "$type": PEER, "iroh": id, "channels": channels, "lease": lease, "createdAt": run.args.now });
                Handle::new()
                    .call(put_record(&AtUri::new(&me, PEER, &format!("{nick}.{handle}")), record(&s["agent_id"])).id("peer-agent"))
                    .call(put_record(&AtUri::new(&me, PEER, &format!("{you}.{handle}")), record(&s["you_id"])).id("peer-you"))
                    .state(s)
            }
            "peer-agent" | "peer-you" => {
                result.output()?;
                s[format!("{id}-done")] = json!(true);
                if s["peer-agent-done"] != true || s["peer-you-done"] != true {
                    return Ok(Handle::new().state(s));
                }
                let handle = s["handle"].as_str().unwrap_or_default();
                let peers: Vec<String> = s["agents"].as_array().into_iter().flatten().flat_map(|x| [x["subhandle"].as_str().unwrap_or_default().to_string(), format!("{}.{handle}", x["you"].as_str().unwrap_or_default())]).filter(|p| !p.starts_with('.') && p.contains('.')).collect();
                let channels: Vec<&str> = s["channels"].as_array().into_iter().flatten().filter_map(Value::as_str).collect();
                let stdin = format!("NICK={}\nYOU={}\nHANDLE={handle}\nCHANNELS={}\nPEERS={}\nLISTEN={}\n", s["nick"].as_str().unwrap_or_default(), s["you"].as_str().unwrap_or_default(), channels.join(" "), peers.join(" "), s["listen_port"]);
                let shell = s["shell"].as_str().unwrap_or_default().to_string();
                Handle::new().call(exec("start", &shell, START, stdin, 120_000)).state(s)
            }
            "start" => {
                let active = exec_out(&result, "start")?;
                let handle = s["handle"].as_str().unwrap_or_default();
                let nick = s["nick"].as_str().unwrap_or_default().to_string();
                let agent = json!({ "nick": nick, "subhandle": format!("{nick}.{handle}"), "you": s["you"], "vm": s["vm"], "shell": s["shell"], "channels": s["channels"], "listen_port": s["listen_port"], "lease": s["lease"], "expires": s["expires"], "createdAt": run.args.now });
                let mut agents = s["agents"].as_array().cloned().unwrap_or_default();
                agents.push(agent.clone());
                s["agent"] = agent;
                s["active"] = json!(active.trim());
                Handle::new().call(index_put(&me, agents, &run.args.now)).state(s)
            }
            "index-put" => {
                result.output()?;
                let mut out = s["agent"].clone();
                out["units"] = s["active"].clone();
                Handle::new().result(out).done()
            }
            _ => Handle::new(),
        },
        _ => Handle::new(),
    })
}

// list {of?}: the agents an account runs (its computer.socialweb.agents/self).
fn list_handler(run: Run<Args>) -> Result<Handle, String> {
    let of = run.args.of.clone().filter(|s| !s.is_empty()).unwrap_or(run.args.did.clone());
    if !of.starts_with("did:") {
        return Err("of: a DID (or sign in)".into());
    }
    Ok(match run.event {
        Event::Start => Handle::new().call(read("index", &of, INDEX, "self")),
        Event::Result { id, result, .. } if id == "index" => Handle::new().result(json!({ "of": of, "agents": agents_of(&result) })).done(),
        _ => Handle::new(),
    })
}

// stop {nick}: stops its units (if the VM still answers), deletes both peer records, drops it from the list.
fn stop_handler(run: Run<Args>) -> Result<Handle, String> {
    let me = run.args.did.clone();
    if !me.starts_with("did:") {
        return Err("sign in first".into());
    }
    let nick = run.args.nick.clone().unwrap_or_default();
    let mut s = run.state.clone();
    Ok(match run.event {
        Event::Start => Handle::new().call(read("index", &me, INDEX, "self")),
        Event::Result { id, result, .. } => match id.as_str() {
            "index" => {
                let agents = agents_of(&result);
                let agent = agents.iter().find(|x| x["nick"] == nick.as_str()).cloned().ok_or(format!("no agent {nick}"))?;
                let handle = agent["subhandle"].as_str().unwrap_or_default().strip_prefix(&format!("{nick}.")).unwrap_or_default().to_string();
                let you = agent["you"].as_str().unwrap_or_default().to_string();
                let lease = agent["lease"]["uri"].as_str().and_then(|u| u.parse::<AtUri>().ok()).filter(|u| u.repo == me && u.collection == ASSOCIATION);
                s = json!({ "agents": agents.into_iter().filter(|x| x["nick"] != nick.as_str()).collect::<Vec<_>>(), "pending": 3 + lease.is_some() as u64, "stopped": false });
                let mut h = Handle::new()
                    .call(exec("stop", agent["shell"].as_str().unwrap_or_default(), STOP, format!("{nick}\n{you}\n"), 60_000))
                    .call(delete_record(&AtUri::new(&me, PEER, &format!("{nick}.{handle}"))).id("del-agent"))
                    .call(delete_record(&AtUri::new(&me, PEER, &format!("{you}.{handle}"))).id("del-you"));
                // each call its own id: the host runs one call per id
                if let Some(lease) = lease {
                    h = h.call(delete_record(&lease).id("del-lease"));
                }
                h.state(s)
            }
            "stop" | "del-agent" | "del-you" | "del-lease" => {
                if id == "stop" {
                    s["stopped"] = json!(result.ok());
                }
                let pending = s["pending"].as_u64().unwrap_or(1) - 1;
                s["pending"] = json!(pending);
                if pending > 0 {
                    return Ok(Handle::new().state(s));
                }
                let agents = s["agents"].as_array().cloned().unwrap_or_default();
                Handle::new().call(index_put(&me, agents, &run.args.now)).state(s)
            }
            "index-put" => {
                result.output()?;
                Handle::new().result(json!({ "nick": nick, "stopped": s["stopped"] })).done()
            }
            _ => Handle::new(),
        },
        _ => Handle::new(),
    })
}

entries!(
    launch => launch_handler,
    list => list_handler,
    stop => stop_handler
);

#[cfg(test)]
mod tests {
    use serde_json::{Value, json};

    type H = fn(abi::Run<super::Args>) -> Result<abi::Handle, String>;
    fn step(f: H, args: &Value, event: Value, state: &Value) -> Result<Value, String> {
        let mut input = args.clone();
        input["event"] = event;
        input["state"] = state.clone();
        f(serde_json::from_value(input).unwrap()).map(|h| serde_json::to_value(h).unwrap())
    }
    fn res(id: &str, output: Value) -> Value {
        json!({ "type": "result", "id": id, "ok": true, "output": output })
    }
    const ME: &str = "did:plc:alice";
    const A: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
    const Y: &str = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";

    fn shell() -> String {
        let reference = super::b64(r#"{"receipt":{"uri":"at://did:plc:bidder/com.publicdomainrelay.temp.market.receipt/r1","cid":"c"},"bidder":"did:plc:bidder"}"#);
        format!("webrtc:///{}/shell/mcp", reference.replace('+', "-").replace('/', "_").trim_end_matches('='))
    }

    fn args() -> Value {
        json!({ "did": ME, "now": "2026-11-10T00:00:00.000Z", "vm": "vm-1", "shell": shell(), "expires": "2026-11-10T04:00:00.000Z", "nick": "Scout", "model": "m", "base_url": "https://api.example.com", "api_key": "sk-1", "atirc_url": "https://app.example/dl/atirc-x86_64", "atirc_sha256": "a".repeat(64) })
    }

    #[test]
    fn channels_are_atircs() {
        assert_eq!(super::channel("aliceoa.bsky.social"), "#aliceoa_bsky_social");
        assert_eq!(super::channel("scout.aliceoa.bsky.social"), "#scout_aliceoa_bsky_social");
        assert_eq!(super::b64("sk-1"), "c2stMQ==");
    }

    #[test]
    fn launch_prepares_writes_both_peer_records_then_starts() {
        let a = args();
        let s = step(super::launch_handler, &a, json!({ "type": "start" }), &Value::Null).unwrap();
        let s = step(super::launch_handler, &a, res("repo", json!({ "handle": "alice.test" })), &s["state"]).unwrap();
        let other = json!({ "nick": "old", "subhandle": "old.alice.test", "you": "you-old" });
        let s = step(super::launch_handler, &a, res("index", json!({ "uri": "u", "cid": "c", "value": { "agents": [other] } })), &s["state"]).unwrap();
        assert_eq!(s["calls"][0]["params"]["rkey"], "r1");
        assert_eq!(s["calls"][0]["service"], "did:plc:bidder");
        let accept = format!("at://{ME}/com.publicdomainrelay.temp.market.accept/acc1");
        let s = step(super::launch_handler, &a, res("receipt", json!({ "value": { "accept": { "uri": accept, "cid": "ca" } } })), &s["state"]).unwrap();
        let lease = &s["calls"][0]["input"];
        assert_eq!(lease["collection"], "com.publicdomainrelay.temp.market.association");
                assert_eq!(lease["rkey"], "requester_associate-a304c5c525080ebfa7b6205c6c31508c-acc1");
        assert_eq!(lease["record"]["subject"], ME);
        assert_eq!(lease["record"]["contract"], json!({ "uri": accept, "cid": "ca" }));
        assert_eq!(lease["record"]["expiresAt"], "2026-11-10T04:00:00.000Z");
        let s = step(super::launch_handler, &a, res("lease", json!({ "uri": "at://lease", "cid": "cl" })), &s["state"]).unwrap();
        let call = &s["calls"][0];
        assert_eq!(call["nsid"], "computer.socialweb.cicd.process.exec");
        let stdin = call["input"]["stdin"].as_str().unwrap();
        assert!(stdin.contains("NICK=scout\nYOU=you-scout\n") && stdin.contains("API_KEY_B64=c2stMQ==\n"));
        assert!(!call["input"]["command"].as_str().unwrap().contains("sk-1"), "no secret in the command");
        let s = step(super::launch_handler, &a, res("prepare", json!({ "exitCode": 0, "stdout": format!("{{\"agent\":\"{A}\",\"you\":\"{Y}\"}}\n") })), &s["state"]).unwrap();
        assert_eq!(s["calls"][0]["input"]["rkey"], "scout.alice.test");
        assert_eq!(s["calls"][0]["input"]["record"]["iroh"], A);
        assert_eq!(s["calls"][0]["input"]["record"]["channels"], json!(["#alice_test", "#scout_alice_test"]));
        assert_eq!(s["calls"][0]["input"]["record"]["lease"], json!({ "uri": "at://lease", "cid": "cl" }));
        assert_eq!(s["calls"][1]["input"]["rkey"], "you-scout.alice.test");
        let s = step(super::launch_handler, &a, res("peer-agent", json!({})), &s["state"]).unwrap();
        assert!(s["calls"].as_array().is_none_or(|c| c.is_empty()));
        let s = step(super::launch_handler, &a, res("peer-you", json!({})), &s["state"]).unwrap();
        let stdin = s["calls"][0]["input"]["stdin"].as_str().unwrap();
        assert!(stdin.contains("CHANNELS=#alice_test #scout_alice_test\n") && stdin.contains("PEERS=old.alice.test you-old.alice.test\n") && stdin.contains("LISTEN=6680\n"));
        let s = step(super::launch_handler, &a, res("start", json!({ "exitCode": 0, "stdout": "active active active " })), &s["state"]).unwrap();
        assert_eq!(s["calls"][0]["input"]["record"]["agents"][1]["subhandle"], "scout.alice.test");
        let done = step(super::launch_handler, &a, res("index-put", json!({})), &s["state"]).unwrap();
        assert_eq!(done["result"]["you"], "you-scout");
        assert_eq!(done["result"]["units"], "active active active");
    }

    #[test]
    fn launch_refuses_bad_input() {
        for (k, v) in [("nick", "you-x"), ("api_key", "a\"b"), ("base_url", "ftp://x"), ("shell", "https://x"), ("shell", "webrtc:///abc/shell/mcp"), ("expires", "soon")] {
            let mut a = args();
            a[k] = json!(v);
            assert!(step(super::launch_handler, &a, json!({ "type": "start" }), &Value::Null).is_err(), "{k}");
        }
    }

    #[test]
    fn stop_deletes_both_records_even_if_the_vm_is_gone() {
        let a = json!({ "did": ME, "now": "n", "nick": "scout" });
        let lease = format!("at://{ME}/com.publicdomainrelay.temp.market.association/requester_associate-x-acc1");
        let agent = json!({ "nick": "scout", "subhandle": "scout.alice.test", "you": "you-scout", "shell": "webrtc:///abc/shell/mcp", "lease": { "uri": lease, "cid": "cl" } });
        let s = step(super::stop_handler, &a, res("index", json!({ "uri": "u", "cid": "c", "value": { "agents": [agent] } })), &Value::Null).unwrap();
        assert_eq!(s["calls"][1]["input"]["rkey"], "scout.alice.test");
        assert_eq!(s["calls"][2]["input"]["rkey"], "you-scout.alice.test");
        assert_eq!(s["calls"][3]["input"]["collection"], "com.publicdomainrelay.temp.market.association");
        let ids: Vec<&str> = s["calls"].as_array().unwrap().iter().map(|c| c["id"].as_str().unwrap()).collect();
        assert_eq!(ids, ["stop", "del-agent", "del-you", "del-lease"], "one call per id");
        let s = step(super::stop_handler, &a, json!({ "type": "result", "id": "stop", "ok": false, "error": "gone" }), &s["state"]).unwrap();
        let s = step(super::stop_handler, &a, res("del-agent", json!({})), &s["state"]).unwrap();
        let s = step(super::stop_handler, &a, res("del-you", json!({})), &s["state"]).unwrap();
        let s = step(super::stop_handler, &a, res("del-lease", json!({})), &s["state"]).unwrap();
        assert_eq!(s["calls"][0]["input"]["record"]["agents"], json!([]));
        let done = step(super::stop_handler, &a, res("index-put", json!({})), &s["state"]).unwrap();
        assert_eq!(done["result"], json!({ "nick": "scout", "stopped": false }));
    }
}