sylphx-cli 0.29.11

//! The `sylphx` binary against a local fake API: generated commands build the
//! wire request (ids, parents from the linked project, flags, update masks,
//! etags), wait for Operations, refuse unconfirmed destructive calls, and
//! `sylphx mcp` speaks MCP over stdio, and `sylphx build run` drives a lease
//! and its guest end to end.

use parking_lot::Mutex;
use std::io::Write;
use std::path::PathBuf;
use std::process::{Command, Stdio};
use std::sync::Arc;

use serde_json::{json, Value};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

#[tokio::test]
async fn apply_offline_refuses_an_undeclarable_kind_without_a_backend() {
    let sb = Sandbox::new("apply-offline-refused");
    std::fs::write(
        sb.dir.join("sylphx.toml"),
        "[[resource]]\nkind = 'hosting.sylphx.com/Service'\nname = 'main'\n",
    )
    .unwrap();
    let out = run(
        &sb,
        "http://127.0.0.1:1",
        &["apply", "--offline", "--detailed-exitcode"],
    )
    .await;
    assert_eq!(out.status.code(), Some(1));
    let err = String::from_utf8_lossy(&out.stderr);
    assert!(err.contains("hosting.sylphx.com/Service"), "{err}");
    assert!(err.contains("is written by `[[services]]`"), "{err}");
}

#[tokio::test]
async fn apply_json_empty_manifest_has_no_changes_and_errors_exit_one() {
    let sb = Sandbox::new("apply-empty");
    std::fs::write(sb.dir.join("declarations.json"), "{}").unwrap();
    let out = run(
        &sb,
        "http://127.0.0.1:1",
        &[
            "apply",
            "--manifest",
            "declarations.json",
            "--detailed-exitcode",
            "-o",
            "json",
        ],
    )
    .await;
    assert_eq!(
        out.status.code(),
        Some(0),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let printed: Value = serde_json::from_slice(&out.stdout).unwrap();
    assert_eq!(printed["resources"], json!([]));
    let out = run(
        &sb,
        "http://127.0.0.1:1",
        &["apply", "--manifest", "missing.json", "--detailed-exitcode"],
    )
    .await;
    assert_eq!(out.status.code(), Some(1));
}

const ENV: &str = "orgs/org_a/projects/prj_a/envs/env_a";

#[derive(Debug, Clone)]
struct Seen {
    line: String,
    head: String,
    body: String,
}

type Log = Arc<Mutex<Vec<Seen>>>;

async fn serve(replies: Vec<(u16, Value)>) -> (String, Log) {
    serve_bytes(
        replies
            .into_iter()
            .map(|(s, v)| (s, "application/json", v.to_string().into_bytes()))
            .collect(),
    )
    .await
}

/// Like [`serve`], with each reply's content type and raw body.
async fn serve_bytes(replies: Vec<(u16, &'static str, Vec<u8>)>) -> (String, Log) {
    serve_bytes_on(TcpListener::bind("127.0.0.1:0").await.unwrap(), replies).await
}

async fn serve_bytes_on(
    listener: TcpListener,
    replies: Vec<(u16, &'static str, Vec<u8>)>,
) -> (String, Log) {
    let url = format!("http://{}", listener.local_addr().unwrap());
    let log: Log = Arc::new(Mutex::new(Vec::new()));
    let seen = log.clone();
    tokio::spawn(async move {
        for (status, content_type, body) in replies {
            let (mut sock, _) = listener.accept().await.unwrap();
            let mut buf = Vec::new();
            let mut chunk = [0u8; 8192];
            let (head, body_in) = loop {
                let n = sock.read(&mut chunk).await.unwrap();
                buf.extend_from_slice(&chunk[..n]);
                let text = String::from_utf8_lossy(&buf).to_string();
                if let Some(i) = text.find("\r\n\r\n") {
                    let head = text[..i].to_string();
                    let len = head
                        .lines()
                        .find_map(|l| {
                            l.to_ascii_lowercase()
                                .strip_prefix("content-length:")
                                .map(|v| v.trim().parse::<usize>().unwrap())
                        })
                        .unwrap_or(0);
                    while buf.len() < i + 4 + len {
                        let n = sock.read(&mut chunk).await.unwrap();
                        buf.extend_from_slice(&chunk[..n]);
                    }
                    break (
                        head,
                        String::from_utf8_lossy(&buf[i + 4..i + 4 + len]).to_string(),
                    );
                }
            };
            let line = head.lines().next().unwrap_or_default().to_string();
            seen.lock().push(Seen {
                line,
                head,
                body: body_in,
            });
            let head = format!(
                "HTTP/1.1 {status} X\r\ncontent-type: {content_type}\r\ncontent-length: {}\r\nconnection: close\r\n\r\n",
                body.len()
            );
            sock.write_all(head.as_bytes()).await.unwrap();
            sock.write_all(&body).await.unwrap();
            sock.shutdown().await.ok();
        }
    });
    (url, log)
}

struct Sandbox {
    dir: PathBuf,
}

impl Sandbox {
    fn new(name: &str) -> Self {
        let dir =
            std::env::temp_dir().join(format!("sylphx-cli-test-{name}-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(dir.join(".sylphx")).unwrap();
        std::fs::write(
            dir.join(".sylphx/project.json"),
            json!({"org": "orgs/org_a", "project": "orgs/org_a/projects/prj_a", "env": ENV})
                .to_string(),
        )
        .unwrap();
        Self { dir }
    }

    fn cmd(&self, url: &str, args: &[&str]) -> Command {
        let mut c = Command::new(env!("CARGO_BIN_EXE_sylphx"));
        c.args(args)
            .current_dir(&self.dir)
            .env("SYLPHX_BASE_URL", url)
            .env("SYLPHX_API_KEY", "sylphx_sk_test")
            .env("SYLPHX_BUILD_CACHE_URL", "http://127.0.0.1:9")
            .env("SYLPHX_CONFIG_DIR", self.dir.join("config"))
            .env("SYLPHX_NO_KEYCHAIN", "1")
            .env("SYLPHX_NO_BROWSER", "1")
            .env_remove("SYLPHX_ENVIRONMENT")
            .stdin(Stdio::null());
        c
    }
}

impl Drop for Sandbox {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.dir);
    }
}

async fn run(sb: &Sandbox, url: &str, args: &[&str]) -> std::process::Output {
    let mut c = sb.cmd(url, args);
    tokio::task::spawn_blocking(move || c.output().unwrap())
        .await
        .unwrap()
}

#[tokio::test(flavor = "multi_thread")]
async fn a_link_file_in_another_shape_never_shadows_the_key_scope() {
    // The older `{orgId, projectId}` file names no org, project or env: it is
    // skipped with a warning, and the key's own env (whoami) is the default.
    let (url, log) = serve(vec![
        (200, whoami_reply()),
        (
            200,
            json!({"name": format!("{ENV}/databases"), "databases": []}),
        ),
    ])
    .await;
    let sb = Sandbox::new("legacy-link");
    std::fs::write(
        sb.dir.join(".sylphx/project.json"),
        json!({"projectId": "proj_x", "orgId": "org_x"}).to_string(),
    )
    .unwrap();
    let out = run(&sb, &url, &["data", "databases", "list", "-o", "json"]).await;
    let err = String::from_utf8_lossy(&out.stderr);
    assert!(out.status.success(), "{err}");
    assert!(
        err.contains("warning: ignoring") && err.contains(".sylphx/project.json"),
        "{err}"
    );
    let log = log.lock().clone();
    assert!(
        log[0].line.starts_with("GET /v1/whoami "),
        "{}",
        log[0].line
    );
    assert!(
        log[1].line.starts_with(&format!("GET /v1/{ENV}/databases")),
        "{}",
        log[1].line
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn create_takes_the_id_and_the_linked_env_and_waits() {
    let db = json!({"name": format!("{ENV}/databases/main"), "spec": {"compute_units": 2}});
    let (url, log) = serve(vec![
        (200, json!({"name": format!("{ENV}/operations/op_1"), "done": false})),
        (200, json!({"name": format!("{ENV}/operations/op_1"), "done": true,
                     "response": {"@type": "type.googleapis.com/sylphx.data.v1.Database", "name": format!("{ENV}/databases/main"), "spec": {"compute_units": 2}}})),
    ])
    .await;
    let sb = Sandbox::new("create");
    let out = run(
        &sb,
        &url,
        &[
            "data",
            "databases",
            "create",
            "main",
            "--spec.compute-units",
            "2",
            "-o",
            "json",
        ],
    )
    .await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let printed: Value = serde_json::from_slice(&out.stdout).unwrap();
    assert_eq!(printed, db);
    let log = log.lock().clone();
    assert!(
        log[0]
            .line
            .starts_with(&format!("POST /v1/{ENV}/databases?database_id=main ")),
        "{}",
        log[0].line
    );
    assert_eq!(
        serde_json::from_str::<Value>(&log[0].body).unwrap(),
        json!({"spec": {"compute_units": 2}})
    );
    assert!(log[0]
        .head
        .to_ascii_lowercase()
        .contains("idempotency-key:"));
    assert!(log[1]
        .line
        .starts_with(&format!("POST /v1/{ENV}/operations/op_1:wait?timeout=60s ")));
}

#[tokio::test(flavor = "multi_thread")]
async fn update_reads_the_etag_and_sends_a_mask_of_the_set_flags() {
    let key = format!("{ENV}/api_keys/key_a");
    let (url, log) = serve(vec![
        (200, json!({"name": key, "meta": {"etag": "\"k3\""}})),
        (200, json!({"name": key, "spec": {"label": "ci"}})),
    ])
    .await;
    let sb = Sandbox::new("update");
    let out = run(
        &sb,
        &url,
        &[
            "access",
            "api-keys",
            "update",
            "key_a",
            "--spec.label",
            "ci",
            "-o",
            "name",
        ],
    )
    .await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    assert_eq!(String::from_utf8_lossy(&out.stdout).trim(), key);
    let log = log.lock().clone();
    assert!(log[0].line.starts_with(&format!("GET /v1/{key} ")));
    assert!(
        log[1]
            .line
            .starts_with(&format!("PATCH /v1/{key}?update_mask=spec.label ")),
        "{}",
        log[1].line
    );
    assert_eq!(
        serde_json::from_str::<Value>(&log[1].body).unwrap(),
        json!({"name": key, "meta": {"etag": "\"k3\""}, "spec": {"label": "ci"}})
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn a_destructive_call_needs_yes_without_a_terminal() {
    let (url, log) = serve(vec![(
        200,
        json!({"name": format!("{ENV}/api_keys/key_a"), "meta": {"etag": "\"e\""}}),
    )])
    .await;
    let sb = Sandbox::new("destructive");
    let out = run(&sb, &url, &["access", "api-keys", "revoke", "key_a"]).await;
    assert_eq!(out.status.code(), Some(2));
    assert!(String::from_utf8_lossy(&out.stderr).contains("--yes"));
    assert!(
        log.lock().iter().all(|s| !s.line.starts_with("POST")),
        "nothing was revoked"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn an_api_error_prints_the_problem_and_exits_1() {
    let (url, _log) = serve(vec![(
        403,
        json!({"code": "PERMISSION_DENIED", "status": 403, "detail": "missing scope data:read", "retryable": false, "instance": "req_9"}),
    )])
    .await;
    let sb = Sandbox::new("error");
    let out = run(&sb, &url, &["data", "databases", "list"]).await;
    assert_eq!(out.status.code(), Some(1));
    let err = String::from_utf8_lossy(&out.stderr);
    assert!(
        err.contains("PERMISSION_DENIED (403): missing scope data:read"),
        "{err}"
    );
    assert!(err.contains("req_9"));
}

#[tokio::test(flavor = "multi_thread")]
async fn list_defaults_to_the_linked_env_and_prints_a_table() {
    let (url, log) = serve(vec![(
        200,
        json!({"databases": [{"name": format!("{ENV}/databases/main"), "status": {"conditions": [{"type": "Ready", "status": "TRUE"}]}}]}),
    )])
    .await;
    let sb = Sandbox::new("list");
    let out = run(
        &sb,
        &url,
        &["data", "databases", "list", "--page-size", "10"],
    )
    .await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let text = String::from_utf8_lossy(&out.stdout);
    assert!(text.starts_with("NAME"), "{text}");
    assert!(text.contains("databases/main  true"));
    assert!(log.lock()[0]
        .line
        .starts_with(&format!("GET /v1/{ENV}/databases?page_size=10 ")));
}

/// `sylphx money revenue summary` reads the linked org's revenue summary
/// (the org-level singleton `orgs/{org}/revenue`): `--by` is the split.
#[tokio::test(flavor = "multi_thread")]
async fn revenue_summary_reads_the_linked_orgs_revenue() {
    let (url, log) = serve(vec![(
        200,
        json!({"name": "orgs/org_a/revenue", "currency_code": "USD", "total": "1500", "split": "source"}),
    )])
    .await;
    let sb = Sandbox::new("revenue");
    let out = run(
        &sb,
        &url,
        &[
            "money", "revenue", "summary", "--period", "mtd", "--by", "source", "--output", "json",
        ],
    )
    .await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    assert!(String::from_utf8_lossy(&out.stdout).contains("1500"));
    let line = log.lock()[0].line.clone();
    assert!(line.starts_with("GET /v1/orgs/org_a/revenue?"), "{line}");
    assert!(
        line.contains("period=mtd") && line.contains("split=source"),
        "{line}"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn sylphx_env_overrides_the_link_file() {
    let other = "orgs/org_b/projects/prj_b/envs/env_b";
    let (url, log) = serve(vec![(200, json!({"databases": []}))]).await;
    let sb = Sandbox::new("env-override");
    let mut c = sb.cmd(&url, &["data", "databases", "list"]);
    c.env("SYLPHX_ENVIRONMENT", other);
    let out = tokio::task::spawn_blocking(move || c.output().unwrap())
        .await
        .unwrap();
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let seen = log.lock();
    assert!(
        seen[0]
            .line
            .starts_with(&format!("GET /v1/{other}/databases")),
        "{}",
        seen[0].line
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn sylphx_env_of_another_shape_is_a_usage_error_and_calls_nothing() {
    let (url, log) = serve(vec![]).await;
    let sb = Sandbox::new("env-bad");
    for bad in [
        "orgs/org_b/projects/prj_b",
        "env_b",
        "orgs//projects/p/envs/e",
        "orgs/o/projects/p/envs/e/x",
    ] {
        let mut c = sb.cmd(&url, &["data", "databases", "list"]);
        c.env("SYLPHX_ENVIRONMENT", bad);
        let out = tokio::task::spawn_blocking(move || c.output().unwrap())
            .await
            .unwrap();
        assert!(!out.status.success(), "{bad}");
        let err = String::from_utf8_lossy(&out.stderr);
        assert!(
            err.contains("SYLPHX_ENVIRONMENT must be a full environment name"),
            "{bad}: {err}"
        );
    }
    assert!(log.lock().is_empty());
}

#[tokio::test(flavor = "multi_thread")]
async fn login_with_an_empty_stdin_starts_the_device_flow() {
    let (url, log) = serve(vec![
        (
            200,
            json!({"device_code": "dc", "user_code": "BCDF-GHJK", "verification_uri": "https://sylphx.com/device",
                   "verification_uri_complete": "https://sylphx.com/device?user_code=BCDF-GHJK",
                   "interval": 1, "expires_in": 600}),
        ),
        (400, json!({"error": "access_denied"})),
    ])
    .await;
    let sb = Sandbox::new("login-stdin");
    let mut c = sb.cmd(&url, &["login", "--org", "acme"]);
    c.env_remove("SYLPHX_API_KEY");
    let out = tokio::task::spawn_blocking(move || c.output().unwrap())
        .await
        .unwrap();
    let err = String::from_utf8_lossy(&out.stderr);
    assert!(!out.status.success());
    assert!(
        err.contains("https://sylphx.com/device?user_code=BCDF-GHJK"),
        "{err}"
    );
    assert!(err.contains("denied"), "{err}");
    let seen = log.lock();
    assert!(seen[0]
        .line
        .starts_with("POST /v1/access/device/authorize "));
    assert!(
        seen[0].body.contains("\"org\":\"acme\""),
        "{}",
        seen[0].body
    );
    assert!(seen[1].line.starts_with("POST /v1/access/device/token "));
}

#[tokio::test(flavor = "multi_thread")]
async fn login_keeps_polling_through_a_transient_server_error() {
    let (url, log) = serve(vec![
        (
            200,
            json!({"device_code": "dc", "user_code": "BCDF-GHJK", "verification_uri": "https://sylphx.com/device",
                   "interval": 1, "expires_in": 600}),
        ),
        (503, json!({"code": "UNAVAILABLE"})),
        (
            200,
            json!({"keys": [{"api_key": "sylphx_sk_test_issued", "org": {"id": "org_a", "slug": "acme"}, "key": {"name": "orgs/org_a/api_keys/key_cli"}}], "active_after_ms": 300}),
        ),
    ])
    .await;
    let sb = Sandbox::new("login-503");
    let mut c = sb.cmd(&url, &["login", "--org", "acme", "--no-verify"]);
    c.env_remove("SYLPHX_API_KEY");
    let out = tokio::task::spawn_blocking(move || c.output().unwrap())
        .await
        .unwrap();
    let err = String::from_utf8_lossy(&out.stderr);
    assert!(out.status.success(), "{err}");
    assert!(err.contains("Approved in acme"), "{err}");
    let seen = log.lock();
    assert_eq!(seen.len(), 3);
    assert!(seen[1].line.starts_with("POST /v1/access/device/token "));
    assert!(seen[2].line.starts_with("POST /v1/access/device/token "));
}

fn device_replies() -> Vec<(u16, Value)> {
    vec![
        (
            200,
            json!({"device_code": "dc", "user_code": "BCDF-GHJK", "verification_uri": "https://sylphx.com/device",
                   "interval": 1, "expires_in": 600}),
        ),
        (
            200,
            json!({"keys": [{"api_key": "sylphx_sk_test_issued", "org": {"id": "org_a", "slug": "acme"}, "key": {"name": "orgs/org_a/api_keys/key_cli"}}], "active_after_ms": 300}),
        ),
    ]
}

fn two_org_replies() -> Vec<(u16, Value)> {
    let mut replies = device_replies();
    replies[1].1 = json!({"keys": [
        {"api_key": "sylphx_sk_org_a", "key": {"name": "orgs/org_a/api_keys/cli_a"}, "org": {"id": "org_a", "slug": "acme"}},
        {"api_key": "sylphx_sk_org_b", "key": {"name": "orgs/org_b/api_keys/cli_b"}, "org": {"id": "org_b", "slug": "beta"}}
    ], "active_after_ms": 0});
    replies
}

async fn run_stored(sb: &Sandbox, url: &str, args: &[&str]) -> std::process::Output {
    let mut c = sb.cmd(url, args);
    c.env_remove("SYLPHX_API_KEY");
    tokio::task::spawn_blocking(move || c.output().unwrap())
        .await
        .unwrap()
}

fn assert_success(out: &std::process::Output) {
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn device_login_stores_all_keys_and_selects_org_per_command() {
    let mut replies = two_org_replies();
    replies.extend((0..7).map(|_| (200, whoami_reply())));
    replies.push((
        200,
        json!({"name": "orgs/org_b/projects/prj_b/envs/env_b/databases/db_x"}),
    ));
    let (url, log) = serve(replies).await;
    let sb = Sandbox::new("login-two-orgs");
    let login = run_stored(&sb, &url, &["login"]).await;
    assert_success(&login);
    for stream in [&login.stdout, &login.stderr] {
        assert!(
            !String::from_utf8_lossy(stream).contains("sylphx_sk_org_"),
            "a login secret was printed"
        );
    }
    let creds: Value = serde_json::from_str(
        &std::fs::read_to_string(sb.dir.join("config/credentials.json")).unwrap(),
    )
    .unwrap();
    assert_eq!(creds["orgs"].as_array().unwrap().len(), 2);
    assert_eq!(creds["orgs"][0]["api_key"], "sylphx_sk_org_a");
    assert_eq!(creds["orgs"][1]["api_key"], "sylphx_sk_org_b");
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        assert_eq!(
            std::fs::metadata(sb.dir.join("config/credentials.json"))
                .unwrap()
                .permissions()
                .mode()
                & 0o777,
            0o600
        );
    }
    assert_success(&run_stored(&sb, &url, &["whoami", "--org", "beta"]).await);
    assert_success(&run_stored(&sb, &url, &["whoami"]).await);
    let mut c = sb.cmd(&url, &["whoami"]);
    c.env_remove("SYLPHX_API_KEY")
        .env("SYLPHX_ENVIRONMENT", "orgs/org_b/projects/prj_b/envs/env_b");
    assert_success(
        &tokio::task::spawn_blocking(move || c.output().unwrap())
            .await
            .unwrap(),
    );
    std::fs::remove_file(sb.dir.join(".sylphx/project.json")).unwrap();
    assert_success(&run_stored(&sb, &url, &["org", "use", "beta"]).await);
    assert_success(&run_stored(&sb, &url, &["whoami"]).await);
    assert_success(&run_stored(&sb, &url, &["whoami", "--org", "orgs/org_a"]).await);
    assert_success(
        &run_stored(
            &sb,
            &url,
            &[
                "data",
                "databases",
                "get",
                "orgs/org_b/projects/prj_b/envs/env_b/databases/db_x",
            ],
        )
        .await,
    );
    let seen = log.lock();
    let body: Value = serde_json::from_str(&seen[0].body).unwrap();
    assert!(
        body.get("org").is_none(),
        "login without --org must not lock selection"
    );
    for (request, key) in seen[2..].iter().zip([
        "sylphx_sk_org_a",
        "sylphx_sk_org_b",
        "sylphx_sk_org_b",
        "sylphx_sk_org_a",
        "sylphx_sk_org_b",
        "sylphx_sk_org_b",
        "sylphx_sk_org_a",
        "sylphx_sk_org_b",
    ]) {
        assert!(
            request
                .head
                .to_ascii_lowercase()
                .contains(&format!("authorization: bearer {key}")),
            "{}",
            request.head
        );
    }
    assert_eq!(seen.len(), 10);
}

#[tokio::test(flavor = "multi_thread")]
async fn fixed_org_relogin_replaces_only_that_org_credential() {
    let mut replies = two_org_replies();
    replies.extend(device_replies());
    let (url, log) = serve(replies).await;
    let sb = Sandbox::new("relogin-one-org");
    assert_success(&run_stored(&sb, &url, &["login", "--no-verify"]).await);
    assert_success(&run_stored(&sb, &url, &["login", "--org", "acme", "--no-verify"]).await);
    let creds: Value = serde_json::from_str(
        &std::fs::read_to_string(sb.dir.join("config/credentials.json")).unwrap(),
    )
    .unwrap();
    assert_eq!(creds["orgs"].as_array().unwrap().len(), 2);
    assert_eq!(creds["orgs"][0]["api_key"], "sylphx_sk_test_issued");
    assert_eq!(creds["orgs"][1]["api_key"], "sylphx_sk_org_b");
    let seen = log.lock();
    let requested: Value = serde_json::from_str(&seen[2].body).unwrap();
    assert_eq!(requested["org"], "acme");
}

#[tokio::test(flavor = "multi_thread")]
async fn logout_selected_org_preserves_other_keys_and_all_revokes_the_rest() {
    let mut replies = two_org_replies();
    replies.extend([
        (200, json!({"ok": true})),
        (200, whoami_reply()),
        (200, json!({"ok": true})),
    ]);
    let (url, log) = serve(replies).await;
    let sb = Sandbox::new("logout-two-orgs");
    assert_success(&run_stored(&sb, &url, &["login", "--no-verify"]).await);
    assert_success(&run_stored(&sb, &url, &["logout", "--org", "acme"]).await);
    let path = sb.dir.join("config/credentials.json");
    let creds: Value = serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
    assert_eq!(creds["orgs"].as_array().unwrap().len(), 1);
    assert_eq!(creds["orgs"][0]["api_key"], "sylphx_sk_org_b");
    assert_eq!(creds["active_org"], "org_b");
    assert_success(&run_stored(&sb, &url, &["whoami", "--org", "beta"]).await);
    assert_success(&run_stored(&sb, &url, &["logout", "--all"]).await);
    assert!(!path.exists());
    let seen = log.lock();
    assert!(seen[2]
        .line
        .starts_with("POST /v1/access/keys/self:revoke "));
    assert!(seen[2].head.contains("Bearer sylphx_sk_org_a"));
    assert!(seen[3].head.contains("Bearer sylphx_sk_org_b"));
    assert!(seen[4]
        .line
        .starts_with("POST /v1/access/keys/self:revoke "));
    assert!(seen[4].head.contains("Bearer sylphx_sk_org_b"));
}

#[tokio::test(flavor = "multi_thread")]
async fn supplied_login_and_logout_preserve_scoped_org_credentials() {
    let mut replies = two_org_replies();
    replies.extend([
        (200, whoami_reply()),
        (200, whoami_reply()),
        (200, json!({"ok": true})),
        (200, json!({"ok": true})),
        (200, json!({"ok": true})),
    ]);
    let (url, log) = serve(replies).await;
    let sb = Sandbox::new("supplied-login-orgs");
    assert_success(&run_stored(&sb, &url, &["login", "--no-verify"]).await);
    assert_success(
        &run_stored(
            &sb,
            &url,
            &["login", "--key", "sylphx_sk_supplied", "--no-verify"],
        )
        .await,
    );
    assert_success(&run_stored(&sb, &url, &["whoami"]).await);
    assert_success(&run_stored(&sb, &url, &["whoami", "--org", "beta"]).await);
    assert_success(&run_stored(&sb, &url, &["logout"]).await);
    let path = sb.dir.join("config/credentials.json");
    let creds: Value = serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
    assert_eq!(creds["orgs"].as_array().unwrap().len(), 2);
    assert!(creds.get("api_key").is_none());
    assert_success(&run_stored(&sb, &url, &["logout", "--all"]).await);
    assert!(!path.exists());
    let seen = log.lock();
    for (request, key) in seen[2..].iter().zip([
        "sylphx_sk_supplied",
        "sylphx_sk_org_b",
        "sylphx_sk_supplied",
        "sylphx_sk_org_a",
        "sylphx_sk_org_b",
    ]) {
        assert!(request.head.contains(&format!("Bearer {key}")));
    }
    assert_eq!(seen.len(), 7);
}

#[tokio::test(flavor = "multi_thread")]
async fn explicit_credentials_override_org_and_legacy_file_credentials_still_work() {
    let (url, log) = serve(vec![(200, whoami_reply()); 3]).await;
    let sb = Sandbox::new("legacy-org-credentials");
    std::fs::create_dir_all(sb.dir.join("config")).unwrap();
    let path = sb.dir.join("config/credentials.json");
    std::fs::write(&path, json!({"api_key": "sylphx_sk_legacy", "store": "file", "base_url": url, "key_name": "orgs/org_a/api_keys/old"}).to_string()).unwrap();
    assert_success(&run_stored(&sb, &url, &["whoami"]).await);
    let mut creds: Value = serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
    creds["orgs"] = json!([{"id": "org_b", "slug": "beta", "store": "file", "key_name": "orgs/org_b/api_keys/cli", "api_key": "sylphx_sk_org_b"}]);
    creds["active_org"] = json!("org_b");
    std::fs::write(&path, creds.to_string()).unwrap();
    assert_success(&run(&sb, &url, &["whoami", "--org", "beta"]).await);
    assert_success(
        &run(
            &sb,
            &url,
            &["whoami", "--org", "beta", "--key", "sylphx_sk_explicit"],
        )
        .await,
    );
    let seen = log.lock();
    assert!(seen[0].head.contains("Bearer sylphx_sk_legacy"));
    assert!(seen[1].head.contains("Bearer sylphx_sk_test"));
    assert!(seen[2].head.contains("Bearer sylphx_sk_explicit"));
}

#[tokio::test(flavor = "multi_thread")]
async fn missing_org_and_rejected_supplied_login_never_discard_other_credentials() {
    let mut replies = two_org_replies();
    replies.push((401, json!({"code": "UNAUTHENTICATED", "status": 401, "detail": "invalid credential", "retryable": false})));
    let (url, log) = serve(replies).await;
    let sb = Sandbox::new("org-credentials-retained");
    assert_success(&run_stored(&sb, &url, &["login", "--no-verify"]).await);
    let path = sb.dir.join("config/credentials.json");
    let before = std::fs::read(&path).unwrap();
    assert!(!run_stored(&sb, &url, &["whoami", "--org", "absent"])
        .await
        .status
        .success());
    assert!(!run_stored(&sb, &url, &["org", "use", "absent"])
        .await
        .status
        .success());
    assert!(!run_stored(&sb, &url, &["logout", "--org", "absent"])
        .await
        .status
        .success());
    assert!(
        !run_stored(&sb, &url, &["login", "--api-key", "sylphx_sk_invalid"])
            .await
            .status
            .success()
    );
    assert_eq!(std::fs::read(&path).unwrap(), before);
    assert_eq!(log.lock().len(), 3);
}

fn not_yet() -> (u16, Value) {
    (
        401,
        json!({"ok": false, "error": "key_not_yet_propagated", "code": "key_not_yet_propagated", "status": 401}),
    )
}

#[tokio::test(flavor = "multi_thread")]
async fn login_waits_for_a_fresh_key_to_propagate() {
    let mut replies = device_replies();
    // Not yet propagated, then (window passed, replica not refreshed)
    // unknown_key, then verified.
    replies.extend([
        not_yet(),
        (
            401,
            json!({"ok": false, "error": "unknown_key", "code": "unknown_key", "status": 401}),
        ),
        (200, json!({"principal": "principal_kyle"})),
    ]);
    let (url, log) = serve(replies).await;
    let sb = Sandbox::new("login-propagate");
    let mut c = sb.cmd(&url, &["login", "--org", "acme"]);
    c.env_remove("SYLPHX_API_KEY")
        .env_remove("DBUS_SESSION_BUS_ADDRESS");
    let out = tokio::task::spawn_blocking(move || c.output().unwrap())
        .await
        .unwrap();
    let stdout = String::from_utf8_lossy(&out.stdout);
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    assert!(stdout.contains("Signed in as principal_kyle"), "{stdout}");
    assert_eq!(log.lock().len(), 5);
    let creds = std::fs::read_to_string(sb.dir.join("config/credentials.json")).unwrap_or_default();
    assert!(creds.contains("sylphx_sk_test_issued"), "{creds}");
}

#[tokio::test(flavor = "multi_thread")]
async fn login_keeps_an_issued_key_it_cannot_confirm_yet() {
    let mut replies = device_replies();
    replies.extend(std::iter::repeat_with(not_yet).take(6));
    let (url, _log) = serve(replies).await;
    let sb = Sandbox::new("login-unconfirmed");
    let mut c = sb.cmd(&url, &["login", "--org", "acme"]);
    c.env_remove("SYLPHX_API_KEY")
        .env_remove("DBUS_SESSION_BUS_ADDRESS");
    let out = tokio::task::spawn_blocking(move || c.output().unwrap())
        .await
        .unwrap();
    let err = String::from_utf8_lossy(&out.stderr);
    assert!(out.status.success(), "{err}");
    assert!(err.contains("no need to sign in again"), "{err}");
    let creds = std::fs::read_to_string(sb.dir.join("config/credentials.json")).unwrap_or_default();
    assert!(creds.contains("sylphx_sk_test_issued"), "{creds}");
}

/// No keychain and no directory of its own: the shared default is refused,
/// before any key is read or any device grant is started.
#[tokio::test(flavor = "multi_thread")]
async fn login_refuses_the_shared_default_store_without_a_keychain() {
    let (url, log) = serve(vec![]).await;
    let sb = Sandbox::new("login-shared-default");
    let home = sb.dir.join("home");
    let shared = home.join(".config/sylphx/credentials.json");
    for args in [
        vec!["login", "--api-key", "sylphx_sk_test_given", "--no-verify"],
        vec!["login", "--org", "acme"],
    ] {
        let mut c = sb.cmd(&url, &args);
        c.env_remove("SYLPHX_API_KEY")
            .env_remove("SYLPHX_CONFIG_DIR")
            .env_remove("XDG_CONFIG_HOME")
            .env("HOME", &home)
            .env("SYLPHX_NO_KEYCHAIN", "1");
        let out = tokio::task::spawn_blocking(move || c.output().unwrap())
            .await
            .unwrap();
        let err = String::from_utf8_lossy(&out.stderr);
        assert_eq!(out.status.code(), Some(2), "{args:?}: {err}");
        assert!(
            err.contains("--store-file") && err.contains("SYLPHX_CONFIG_DIR"),
            "{err}"
        );
    }
    assert!(!shared.exists(), "nothing written to the shared default");
    assert!(log.lock().is_empty(), "no device grant started");
}

/// `--store-file` keeps the key in the default directory; `SYLPHX_CONFIG_DIR`
/// keeps it in the chosen one.
#[tokio::test(flavor = "multi_thread")]
async fn login_stores_a_file_only_where_asked() {
    let (url, _log) = serve(vec![]).await;
    let sb = Sandbox::new("login-store-file");
    let home = sb.dir.join("home");
    let mut c = sb.cmd(
        &url,
        &[
            "login",
            "--api-key",
            "sylphx_sk_test_given",
            "--no-verify",
            "--store-file",
        ],
    );
    c.env_remove("SYLPHX_API_KEY")
        .env_remove("SYLPHX_CONFIG_DIR")
        .env_remove("XDG_CONFIG_HOME")
        .env("HOME", &home)
        .env("SYLPHX_NO_KEYCHAIN", "1");
    let out = tokio::task::spawn_blocking(move || c.output().unwrap())
        .await
        .unwrap();
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let shared = std::fs::read_to_string(home.join(".config/sylphx/credentials.json")).unwrap();
    assert!(shared.contains("sylphx_sk_test_given"), "{shared}");

    let own = sb.dir.join("own-config");
    let mut c = sb.cmd(
        &url,
        &["login", "--api-key", "sylphx_sk_test_own", "--no-verify"],
    );
    c.env_remove("SYLPHX_API_KEY")
        .env("SYLPHX_CONFIG_DIR", &own)
        .env("SYLPHX_NO_KEYCHAIN", "1");
    let out = tokio::task::spawn_blocking(move || c.output().unwrap())
        .await
        .unwrap();
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let stored = std::fs::read_to_string(own.join("credentials.json")).unwrap();
    assert!(stored.contains("sylphx_sk_test_own"), "{stored}");
}

#[test]
fn mcp_answers_over_stdio() {
    let sb = Sandbox::new("mcp");
    let mut child = sb
        .cmd("http://127.0.0.1:9", &["mcp"])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .spawn()
        .unwrap();
    {
        let stdin = child.stdin.as_mut().unwrap();
        writeln!(stdin, "{}", json!({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "t", "version": "0"}}})).unwrap();
        writeln!(
            stdin,
            "{}",
            json!({"jsonrpc": "2.0", "method": "notifications/initialized"})
        )
        .unwrap();
        writeln!(
            stdin,
            "{}",
            json!({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
        )
        .unwrap();
    }
    drop(child.stdin.take());
    let out = child.wait_with_output().unwrap();
    let lines: Vec<Value> = String::from_utf8_lossy(&out.stdout)
        .lines()
        .map(|l| serde_json::from_str(l).unwrap())
        .collect();
    assert_eq!(lines.len(), 2);
    assert_eq!(lines[0]["result"]["serverInfo"]["name"], "sylphx");
    assert!(lines[1]["result"]["tools"].as_array().unwrap().len() >= 4);
}

fn token_reply(secret: &str) -> Value {
    json!({"name": format!("{ENV}/api_keys/key_child"), "spec": {"scopes": ["hosting:deploy"]},
           "status": {"secret": secret}, "active_after_ms": 3000})
}

fn whoami_reply() -> Value {
    json!({"principal": "principal_x", "api_key": format!("{ENV}/api_keys/key_self"),
           "org": "orgs/org_a", "project": "orgs/org_a/projects/prj_a", "env": ENV,
           "scopes": ["*:write"]})
}

fn civil(secs: u64) -> String {
    // Enough of RFC 3339 for the tests: the date part via the same algorithm.
    let z = (secs / 86400) as i64 + 719468;
    let era = z.div_euclid(146097);
    let doe = z.rem_euclid(146097);
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = yoe + era * 400 + i64::from(m <= 2);
    let r = secs % 86400;
    format!(
        "{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}Z",
        r / 3600,
        r % 3600 / 60,
        r % 60
    )
}

fn now_secs() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_secs()
}

#[tokio::test(flavor = "multi_thread")]
async fn token_prints_only_a_minted_key_scoped_to_one_scope() {
    let (url, log) = serve(vec![
        (200, whoami_reply()),
        (201, token_reply("sylphx_sk_child_1")),
    ])
    .await;
    let sb = Sandbox::new("token-mint");
    let out = run(&sb, &url, &["token", "--scope", "packages:read"]).await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    assert_eq!(String::from_utf8_lossy(&out.stdout), "sylphx_sk_child_1\n");
    assert!(!String::from_utf8_lossy(&out.stderr).contains("sylphx_sk_child_1"));
    let log = log.lock().clone();
    assert!(log[1]
        .line
        .starts_with(&format!("POST /v1/{ENV}/api_keys ")));
    let body: Value = serde_json::from_str(&log[1].body).unwrap();
    assert_eq!(body["spec"]["scopes"], json!(["packages:read"]));
    assert_eq!(body["spec"]["label"], "token:packages:read");
    let exp = body["spec"]["expire_time"].as_str().unwrap();
    assert!(
        exp.ends_with('Z') && exp > civil(now_secs()).as_str(),
        "{exp}"
    );
    // The login key is used to mint, never printed.
    assert!(log[1].head.contains("sylphx_sk_test"));
}

#[tokio::test(flavor = "multi_thread")]
async fn token_is_cached_until_shortly_before_it_expires() {
    let (url, log) = serve(vec![
        (200, whoami_reply()),
        (201, token_reply("sylphx_sk_child_1")),
    ])
    .await;
    let sb = Sandbox::new("token-cache");
    let first = run(&sb, &url, &["token", "--scope", "hosting:deploy"]).await;
    let second = run(&sb, &url, &["token", "--scope", "hosting:deploy"]).await;
    assert_eq!(first.stdout, second.stdout);
    assert_eq!(log.lock().len(), 2, "the second call made no request");
    let dir = sb.dir.join("config/token-cache");
    let files: Vec<_> = std::fs::read_dir(&dir).unwrap().flatten().collect();
    assert_eq!(files.len(), 1);
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let mode = files[0].metadata().unwrap().permissions().mode() & 0o777;
        assert_eq!(mode, 0o600);
    }
    // A different scope is a different token.
    let (url2, log2) = serve(vec![
        (200, whoami_reply()),
        (201, token_reply("sylphx_sk_child_ai")),
    ])
    .await;
    let other = run(&sb, &url2, &["token", "--scope", "ai:inference"]).await;
    assert_eq!(
        String::from_utf8_lossy(&other.stdout),
        "sylphx_sk_child_ai\n"
    );
    assert_eq!(log2.lock().len(), 2);

    // Within 2 minutes of expiry the cache is not used.
    let path = dir
        .read_dir()
        .unwrap()
        .flatten()
        .find(|f| {
            std::fs::read_to_string(f.path())
                .unwrap()
                .contains("child_1")
        })
        .unwrap()
        .path();
    let soon = json!({"token": "sylphx_sk_child_1", "expires_at": now_secs() + 60});
    std::fs::write(&path, soon.to_string()).unwrap();
    let (url3, log3) = serve(vec![
        (200, whoami_reply()),
        (201, token_reply("sylphx_sk_child_2")),
    ])
    .await;
    let again = run(&sb, &url3, &["token", "--scope", "hosting:deploy"]).await;
    assert_eq!(
        String::from_utf8_lossy(&again.stdout),
        "sylphx_sk_child_2\n"
    );
    assert_eq!(log3.lock().len(), 2);
}

#[tokio::test(flavor = "multi_thread")]
async fn token_refuses_an_unregistered_scope() {
    let (url, _log) = serve(vec![
        (200, whoami_reply()),
        (400, json!({"code": "INVALID_FIELD", "status": 400, "retryable": false, "detail": "scopes: unregistered scope nope:nope"})),
    ])
    .await;
    let sb = Sandbox::new("token-unregistered");
    let out = run(&sb, &url, &["token", "--scope", "nope:nope"]).await;
    assert!(!out.status.success());
    assert!(out.stdout.is_empty());
    let err = String::from_utf8_lossy(&out.stderr);
    assert!(
        err.contains("nope:nope") && err.contains("not a registered scope"),
        "{err}"
    );
    // A malformed scope never reaches the network.
    let bad = run(
        &sb,
        "http://127.0.0.1:9",
        &["token", "--scope", "Not A Scope"],
    )
    .await;
    assert!(!bad.status.success() && bad.stdout.is_empty());
}

#[tokio::test(flavor = "multi_thread")]
async fn token_says_when_signed_out() {
    let sb = Sandbox::new("token-signed-out");
    let mut c = sb.cmd(
        "http://127.0.0.1:9",
        &["token", "--scope", "hosting:deploy"],
    );
    c.env_remove("SYLPHX_API_KEY")
        .env("SYLPHX_NO_KEYCHAIN", "1");
    let out = tokio::task::spawn_blocking(move || c.output().unwrap())
        .await
        .unwrap();
    assert!(!out.status.success());
    assert!(out.stdout.is_empty());
    assert!(String::from_utf8_lossy(&out.stderr).contains("not signed in"));
}

#[tokio::test(flavor = "multi_thread")]
async fn token_names_the_scope_a_login_may_not_grant() {
    let (url, _log) = serve(vec![
        (200, whoami_reply()),
        (403, json!({"code": "PERMISSION_DENIED", "status": 403, "retryable": false, "detail": "the caller cannot grant hosting:deploy"})),
    ])
    .await;
    let sb = Sandbox::new("token-denied");
    let out = run(&sb, &url, &["token", "--scope", "hosting:deploy"]).await;
    assert!(!out.status.success());
    assert!(out.stdout.is_empty());
    let err = String::from_utf8_lossy(&out.stderr);
    assert!(err.contains("hosting:deploy"), "{err}");
    assert_eq!(err.trim().lines().count(), 1, "{err}");
}

#[tokio::test(flavor = "multi_thread")]
async fn token_from_an_api_key_that_cannot_mint_prints_it_only_if_short_lived() {
    let key = format!("{ENV}/api_keys/key_self");
    let me = json!({"api_key": key, "org": "orgs/org_a", "project": "orgs/org_a/projects/prj_a",
                    "env": ENV, "scopes": ["hosting:deploy"]});
    let denied = || json!({"code": "PERMISSION_DENIED", "status": 403, "retryable": false, "detail": "missing access:keys:write"});
    // Expires in 30 minutes: an exchange key, handed out as it is.
    let (url, _l) = serve(vec![
        (200, me.clone()),
        (403, denied()),
        (
            200,
            json!({"name": key, "spec": {"expire_time": civil(now_secs() + 1800)}}),
        ),
    ])
    .await;
    let sb = Sandbox::new("token-ci-short");
    let out = run(&sb, &url, &["token", "--scope", "hosting:deploy"]).await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    assert_eq!(String::from_utf8_lossy(&out.stdout), "sylphx_sk_test\n");
    // Expires in a day: refused.
    let (url, _l) = serve(vec![
        (200, me.clone()),
        (403, denied()),
        (
            200,
            json!({"name": key, "spec": {"expire_time": civil(now_secs() + 86400)}}),
        ),
    ])
    .await;
    let sb = Sandbox::new("token-ci-long");
    let out = run(&sb, &url, &["token", "--scope", "hosting:deploy"]).await;
    assert!(!out.status.success());
    assert!(out.stdout.is_empty());
    // Not carrying the scope: refused without reading the key.
    let (url, log) = serve(vec![(200, me), (403, denied())]).await;
    let sb = Sandbox::new("token-ci-other");
    let out = run(&sb, &url, &["token", "--scope", "ai:inference"]).await;
    assert!(!out.status.success() && out.stdout.is_empty());
    assert_eq!(log.lock().len(), 2);
}

/// Enable Auth binds Sylphx Auth into the linked environment's identity slot
/// with the composition route's own request (`PUT .../composition/bindings`),
/// and prints the instance the answer names. The route's server side is
/// covered by the composition bind tests in `sylphx-apps-api`.
#[tokio::test(flavor = "multi_thread")]
async fn auth_enable_binds_the_linked_environment_and_prints_the_instance() {
    let (url, log) = serve(vec![(
        200,
        json!({"status": "active", "providerId": "sylphx-auth",
               "config": {"organizationId": "org-1"}}),
    )])
    .await;
    let sb = Sandbox::new("auth-enable");
    let out = run(&sb, &url, &["auth", "enable"]).await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let text = String::from_utf8_lossy(&out.stdout);
    assert!(text.contains("org-1"), "{text}");
    let log = log.lock().clone();
    assert!(
        log[0]
            .line
            .starts_with("PUT /v1/projects/prj_a/composition/bindings "),
        "{}",
        log[0].line
    );
    assert_eq!(
        serde_json::from_str::<Value>(&log[0].body).unwrap(),
        json!({"capabilityId": "identity", "slot": "identity",
               "providerId": "sylphx-auth", "environmentId": "env_a"})
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn auth_enable_preserves_a_real_cell_bearing_environment_name() {
    let env = "orgs/org_fs01afbdccgn8xdk1ghj5y47w0vtph/projects/prj_fs016wck3akrzep3zkwbv3dt0k9hgr/envs/env_fs01y448d2kqbb8ntphzf61k85wthw";
    let (url, log) = serve(vec![(
        200,
        json!({"status": "active", "providerId": "sylphx-auth",
               "config": {"organizationId": "org-1"}}),
    )])
    .await;
    let sb = Sandbox::new("auth-enable-canonical-env");
    std::fs::write(
        sb.dir.join(".sylphx/project.json"),
        json!({"org": "orgs/org_fs01afbdccgn8xdk1ghj5y47w0vtph",
               "project": "orgs/org_fs01afbdccgn8xdk1ghj5y47w0vtph/projects/prj_fs016wck3akrzep3zkwbv3dt0k9hgr",
               "env": env}).to_string(),
    ).unwrap();
    let out = run(&sb, &url, &["auth", "enable", "--env", env]).await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    assert!(String::from_utf8_lossy(&out.stdout).contains("org-1"));
    let requests = log.lock();
    assert_eq!(requests.len(), 1);
    assert!(requests[0]
        .line
        .starts_with("PUT /v1/projects/prj_fs016wck3akrzep3zkwbv3dt0k9hgr/composition/bindings "));
    assert_eq!(
        serde_json::from_str::<Value>(&requests[0].body).unwrap(),
        json!({"capabilityId": "identity", "slot": "identity",
               "providerId": "sylphx-auth", "environmentId": "env_fs01y448d2kqbb8ntphzf61k85wthw"})
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn auth_status_reads_the_identity_bindings_back() {
    let (url, log) = serve(vec![(
        200,
        json!({"bindings": [
            {"slot": "identity", "environmentId": "e1", "providerId": "sylphx-auth",
             "status": "active", "config": {"organizationId": "org-1"}},
            {"slot": "email", "environmentId": "e1", "providerId": "x"}]}),
    )])
    .await;
    let sb = Sandbox::new("auth-status");
    let out = run(&sb, &url, &["auth", "status", "-o", "json"]).await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let printed: Value = serde_json::from_slice(&out.stdout).unwrap();
    assert_eq!(printed["enabled"], true, "{printed}");
    assert_eq!(printed["bindings"][0]["organizationId"], "org-1");
    assert!(log.lock()[0]
        .line
        .starts_with("GET /v1/projects/prj_a/composition/bindings "));
}

#[tokio::test(flavor = "multi_thread")]
async fn workflows_schedules_read_the_served_compute_route() {
    let sched = json!({"id": "sched_1", "name": "nightly", "status": "active"});
    let (url, log) = serve(vec![
        (200, json!({"schedules": [sched.clone()], "limit": 50})),
        (200, sched),
    ])
    .await;
    let sb = Sandbox::new("schedules");
    let out = run(&sb, &url, &["workflows", "schedules", "list", "-o", "json"]).await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    assert!(String::from_utf8_lossy(&out.stdout).contains("sched_1"));
    let out = run(
        &sb,
        &url,
        &["workflows", "schedules", "get", "sched_1", "-o", "json"],
    )
    .await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let seen = log.lock();
    assert!(
        seen[0].line.starts_with("GET /v1/schedules "),
        "{}",
        seen[0].line
    );
    assert!(
        seen[1].line.starts_with("GET /v1/schedules/sched_1 "),
        "{}",
        seen[1].line
    );
}

/// One Connect envelope of the envd process stream.
fn envd_frame(flags: u8, v: Value) -> Vec<u8> {
    let body = serde_json::to_vec(&v).unwrap();
    let mut out = vec![flags];
    out.extend_from_slice(&(body.len() as u32).to_be_bytes());
    out.extend_from_slice(&body);
    out
}

/// A finished process: optional stdout, then its exit code.
fn envd_process(stdout: &str, code: i64) -> (u16, &'static str, Vec<u8>) {
    use base64::Engine;
    let mut b = envd_frame(0, json!({"event": {"start": {"pid": 7}}}));
    if !stdout.is_empty() {
        let data = base64::engine::general_purpose::STANDARD.encode(stdout);
        b.extend(envd_frame(0, json!({"event": {"data": {"stdout": data}}})));
    }
    b.extend(envd_frame(
        0,
        json!({"event": {"end": {"exitCode": code, "exited": true, "status": format!("exit status {code}")}}}),
    ));
    b.extend(envd_frame(2, json!({})));
    (200, "application/connect+json", b)
}

/// A process that wrote `stderr` and ended with `code`.
fn envd_process_err(stderr: &str, code: i64) -> (u16, &'static str, Vec<u8>) {
    use base64::Engine;
    let data = base64::engine::general_purpose::STANDARD.encode(stderr);
    let mut b = envd_frame(0, json!({"event": {"start": {"pid": 7}}}));
    b.extend(envd_frame(0, json!({"event": {"data": {"stderr": data}}})));
    b.extend(envd_frame(
        0,
        json!({"event": {"end": {"exitCode": code, "exited": true, "status": format!("exit status {code}")}}}),
    ));
    b.extend(envd_frame(2, json!({})));
    (200, "application/connect+json", b)
}

/// A process stream that stops before its end event, as a dropped gateway
/// connection looks to the client.
fn envd_broken() -> (u16, &'static str, Vec<u8>) {
    let b = envd_frame(0, json!({"event": {"start": {"pid": 7}}}));
    (200, "application/connect+json", b)
}

fn git_init(dir: &std::path::Path) {
    let ok = Command::new("git")
        .arg("-C")
        .arg(dir)
        .args(["init", "-q"])
        .status()
        .unwrap()
        .success();
    assert!(ok);
}

/// The whole run against a stub API and guest: a volume-free `build-large`
/// lease, a full sync into its cold owned workspace, the command's output
/// and exit code, and the lease released at the end.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_syncs_runs_and_returns_the_commands_exit_code() {
    let sb = Sandbox::new("build-run");
    git_init(&sb.dir);
    std::fs::write(sb.dir.join("main.rs"), "fn main() {}\n").unwrap();
    let json_reply = |v: Value| (200u16, "application/json", v.to_string().into_bytes());
    let lease = format!("{ENV}/leases/l1");
    // The guest is the same stub: its address is the lease's guest_uri.
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let replies = vec![
        json_reply(
            json!({"name": lease, "spec": {"region": "hk"}, "status": {"state": "ready", "endpoints": {"guestUri": url, "e2bSandboxId": "ABCD1234"}}}),
        ),
        json_reply(json!({"token": "h.p.s", "generation": 1})),
        envd_process("", 0),
        json_reply(json!({})),
        json_reply(json!({})),
        envd_process("", 0),
        envd_process("", 0),
        envd_process("hello from the build machine\n", 3),
        json_reply(json!({"name": lease, "status": {"state": "ending"}})),
    ];
    let (url2, log) = serve_bytes_on(listener, replies).await;
    assert_eq!(url, url2);
    let out = run(
        &sb,
        &url,
        &[
            "build",
            "run",
            "--region",
            "hk",
            "--no-cache",
            "--",
            "cargo",
            "test",
        ],
    )
    .await;
    let err = String::from_utf8_lossy(&out.stderr);
    assert_eq!(out.status.code(), Some(3), "{err}");
    assert_eq!(
        String::from_utf8_lossy(&out.stdout),
        "hello from the build machine\n"
    );
    assert!(
        err.contains("sylphx: synced 2 files") || err.contains("sylphx: synced 1 files"),
        "{err}"
    );
    assert!(err.contains("cold lease-owned workspace"), "{err}");
    assert!(err.contains("sylphx: exit 3 in"), "{err}");
    assert_volume_free_regional_leases(&log, "hk");

    let seen = log.lock();
    let lines: Vec<&str> = seen.iter().map(|s| s.line.as_str()).collect();
    assert!(
        lines[0].starts_with(&format!("POST /v1/{ENV}/leases")),
        "{lines:?}"
    );
    let created: Value = serde_json::from_str(&seen[0].body).unwrap();
    assert_eq!(created["spec"]["shape"], "build-large");
    assert_eq!(created["spec"]["image"], "template:build");
    let net = &created["spec"]["network"];
    assert!(
        net["egress"]
            .as_str()
            .unwrap()
            .eq_ignore_ascii_case("allowlist"),
        "{net}"
    );
    // No internet package host by default: only the build-cache gateway.
    assert_eq!(
        net["allowed_domains"],
        json!(["build-cache.sylphx.net"]),
        "{net}"
    );
    // The guest calls carry the sandbox id and the lease token, never the key.
    let guest: Vec<&Seen> = seen.iter().skip(2).take(6).collect();
    for g in &guest {
        let head = g.head.to_ascii_lowercase();
        assert!(head.contains("e2b-sandbox-id: abcd1234"), "{}", g.head);
        assert!(head.contains("x-access-token: h.p.s"), "{}", g.head);
        assert!(!head.contains("sylphx_sk_test"), "{}", g.head);
    }
    assert!(
        lines[2].starts_with("POST /process.Process/Start"),
        "{lines:?}"
    );
    assert!(
        lines[3].starts_with("POST /files?path=/workspace/.sylphx/in-0000.tar.gz"),
        "{lines:?}"
    );
    assert!(
        lines[4].starts_with("POST /files?path=/workspace/.sylphx/add"),
        "{lines:?}"
    );
    assert!(
        seen[7].body.contains("cargo"),
        "the command is started: {}",
        seen[7].body
    );
    // Cold targets use the scoped remote cache, never persistent local state.
    assert!(
        seen[7].body.contains(r#""/workspace",".","1","cargo""#),
        "{}",
        seen[7].body
    );
    assert!(
        lines[8].contains(":release"),
        "the lease is always released: {lines:?}"
    );
}

/// `-o json` writes NDJSON events, ending with the `result` event.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_json_reports_a_platform_failure_as_125() {
    let sb = Sandbox::new("build-run-json");
    git_init(&sb.dir);
    std::fs::write(sb.dir.join("a.txt"), "a\n").unwrap();
    let (url, _log) = serve(vec![(
        403,
        json!({"code": "PERMISSION_DENIED", "status": 403, "detail": "missing scope sandboxes:read", "retryable": false}),
    )])
    .await;
    let out = run(&sb, &url, &["build", "run", "-o", "json", "--", "true"]).await;
    assert_eq!(out.status.code(), Some(125));
    let stdout = String::from_utf8_lossy(&out.stdout);
    let last: Value = serde_json::from_str(stdout.lines().last().unwrap()).unwrap();
    assert_eq!(last["type"], "result");
    assert_eq!(last["exit_code"], 125);
    assert_eq!(last["outcome"], "platform_error");
    assert_eq!(
        last["retryable"], false,
        "a refused key is not worth a retry"
    );
    assert!(
        last["error"].as_str().unwrap().contains("sandboxes:read"),
        "{last}"
    );
    assert_eq!(last["workspace"], "cold");
}

#[tokio::test(flavor = "multi_thread")]
async fn build_run_usage_errors_exit_2_and_dry_run_creates_nothing() {
    let sb = Sandbox::new("build-run-usage");
    git_init(&sb.dir);
    std::fs::write(sb.dir.join("a.txt"), "abc\n").unwrap();
    // No server: nothing may be called.
    let url = "http://127.0.0.1:9";
    let out = run(&sb, url, &["build", "run", "--size", "huge", "--", "true"]).await;
    assert_eq!(out.status.code(), Some(2));
    let out = run(
        &sb,
        url,
        &["build", "run", "--dry-run", "-o", "json", "--", "true"],
    )
    .await;
    assert_eq!(
        out.status.code(),
        Some(0),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let ev: Value = serde_json::from_str(String::from_utf8_lossy(&out.stdout).trim()).unwrap();
    assert_eq!(ev["type"], "sync");
    assert!(ev["files"].as_u64().unwrap() >= 1, "{ev}");
}

/// `--region gra` pins a volume-free lease even when admission is refused.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_region_pins_the_volume_free_lease() {
    let sb = Sandbox::new("build-run-region");
    git_init(&sb.dir);
    std::fs::write(sb.dir.join("a.txt"), "a\n").unwrap();
    let mut replies = Vec::new();
    // NO_CAPACITY is retryable: the SDK retries it itself before the run
    // gives up, so the stub answers it every time.
    let no_capacity = json!({"code": "NO_CAPACITY", "status": 503, "detail": "No machine can be granted in region `gra` now; retry.", "retryable": true});
    replies.extend(std::iter::repeat_n((503, no_capacity), 8));
    let (url, log) = serve(replies).await;
    let out = run(
        &sb,
        &url,
        &[
            "build",
            "run",
            "-o",
            "json",
            "--region",
            "gra",
            "--queue-timeout",
            "120s",
            "--",
            "true",
        ],
    )
    .await;
    assert_volume_free_regional_leases(&log, "gra");
    // No capacity in the region: the platform exit code, retryable, which the
    // caller can turn into a run in the home region.
    assert_eq!(out.status.code(), Some(125));
    let stdout = String::from_utf8_lossy(&out.stdout);
    let last: Value = serde_json::from_str(stdout.lines().last().unwrap()).unwrap();
    assert_eq!(last["exit_code"], 125);
    assert_eq!(last["retryable"], true, "{last}");
    assert!(last["error"].as_str().unwrap().contains("gra"), "{last}");
}

/// No --region: send an unpinned ephemeral lease directly, without listing,
/// creating or mounting any home-region workspace.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_without_region_requests_automatic_ephemeral_placement() {
    let sb = Sandbox::new("build-run-home");
    git_init(&sb.dir);
    std::fs::write(sb.dir.join("a.txt"), "a\n").unwrap();
    let (url, log) = serve(vec![
        (
            400,
            json!({"code": "SHAPE_NOT_OFFERED", "status": 400, "detail": "shape `build-large` is not offered", "retryable": false}),
        ),
    ])
    .await;
    let out = run(&sb, &url, &["build", "run", "-o", "json", "--", "true"]).await;
    assert_eq!(out.status.code(), Some(125));
    let seen = log.lock();
    assert_eq!(seen.len(), 1, "no workspace API calls");
    assert!(seen[0].line.starts_with(&format!("POST /v1/{ENV}/leases")));
    let lease: Value = serde_json::from_str(&seen[0].body).unwrap();
    assert!(
        lease["spec"]
            .get("region")
            .is_none_or(|r| r.as_str() == Some("")),
        "{lease}"
    );
    assert!(
        lease["spec"]
            .get("volumes")
            .is_none_or(|v| v.as_array().is_some_and(|a| a.is_empty())),
        "automatic placement must not bind a home-region PVC: {lease}"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn build_run_automatic_placement_executes_in_the_selected_fr_cell() {
    let sb = Sandbox::new("build-run-auto-gra");
    git_init(&sb.dir);
    std::fs::write(sb.dir.join("main.rs"), "fn main() {}\n").unwrap();
    let json_reply = |v: Value| (200u16, "application/json", v.to_string().into_bytes());
    let lease = format!("{ENV}/leases/l1");
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let replies = vec![
        json_reply(
            json!({"name": lease, "spec": {"region": "gra"}, "status": {"state": "ready", "endpoints": {"guestUri": url, "e2bSandboxId": "ABCD1234"}}}),
        ),
        json_reply(json!({"token": "h.p.s", "generation": 1})),
        envd_process("", 0),
        json_reply(json!({})),
        json_reply(json!({})),
        envd_process("", 0),
        envd_process("", 0),
        envd_process("automatic FR command\n", 0),
        json_reply(json!({"name": lease, "status": {"state": "ending"}})),
    ];
    let (url, log) = serve_bytes_on(listener, replies).await;
    let out = run(
        &sb,
        &url,
        &[
            "build",
            "run",
            "-o",
            "json",
            "--no-cache",
            "--",
            "cargo",
            "check",
        ],
    )
    .await;
    assert_eq!(
        out.status.code(),
        Some(0),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let stdout = String::from_utf8_lossy(&out.stdout);
    let events: Vec<Value> = stdout
        .lines()
        .filter_map(|l| serde_json::from_str(l).ok())
        .collect();
    let running = events.iter().find(|e| e["type"] == "running").unwrap();
    assert_eq!(running["region"], "gra");
    assert_eq!(running["volume"], false);
    let seen = log.lock();
    assert!(seen[0].line.starts_with(&format!("POST /v1/{ENV}/leases")));
    assert!(!seen.iter().any(|s| s.line.contains("/volumes")));
    let created: Value = serde_json::from_str(&seen[0].body).unwrap();
    assert!(created["spec"]
        .get("region")
        .is_none_or(|r| r.as_str() == Some("")));
    assert!(created["spec"]
        .get("volumes")
        .is_none_or(|v| v.as_array().is_some_and(|a| a.is_empty())));
    assert!(seen[7].body.contains(r#""/workspace",".","1","cargo""#));
    assert!(seen[8].line.contains(":release"));
    for request in seen.iter().skip(2).take(6) {
        let headers = request.head.to_ascii_lowercase();
        assert!(headers.contains("x-access-token: h.p.s"));
        assert!(!headers.contains("sylphx_sk_test"));
    }
}

/// Acceptance fixes the Cell before READY, not only after a machine boots.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_automatic_retry_keeps_the_cell_of_a_lost_pending_lease() {
    automatic_placement_retry("gra", true).await;
}

/// A stream fault during bootstrap may retry the machine, never its placement.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_automatic_retry_keeps_the_cell_after_a_precommand_stream_fault() {
    automatic_placement_retry("gra", false).await;
}

/// Empty canonical region is an accepted home Cell, not unselected placement.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_automatic_retry_pins_a_legacy_empty_accepted_region_to_home() {
    automatic_placement_retry("", true).await;
}

async fn automatic_placement_retry(accepted_region: &str, pending: bool) {
    let sb = Sandbox::new(&format!("auto-retry-{accepted_region}-{pending}"));
    project(&sb.dir);
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let first = format!("{ENV}/leases/first");
    let second = format!("{ENV}/leases/second");
    let retry_region = if accepted_region.is_empty() {
        "hk"
    } else {
        accepted_region
    };
    let json_reply = |v: Value| (200u16, "application/json", v.to_string().into_bytes());
    let mut replies = vec![json_reply(json!({
        "name": first,
        "spec": {"region": accepted_region},
        "status": {"state": if pending { "granted" } else { "ready" },
            "endpoints": {"guestUri": url, "e2bSandboxId": "ABCD1234"}}
    }))];
    if !pending {
        replies.push(json_reply(json!({"token": "h.p.s", "generation": 1})));
        replies.push(envd_broken()); // bootstrap breaks before the user's command
    }
    // No spec in the loss response: affinity must already come from CreateLease.
    replies.push(json_reply(json!({"name": first,
        "status": {"state": "ended", "endReason": "machine_lost"}})));
    replies.extend([
        json_reply(json!({"name": second, "spec": {"region": retry_region},
            "status": {"state": "ready", "endpoints": {"guestUri": url, "e2bSandboxId": "ABCD1234"}}})),
        json_reply(json!({"token": "h.p.t", "generation": 1})),
        envd_process("", 0),
        json_reply(json!({})),
        json_reply(json!({})),
        envd_process("", 0),
        envd_process("", 0),
        envd_process("built once\n", 0),
        json_reply(json!({"name": second, "status": {"state": "ending"}})),
    ]);
    let (_, log) = serve_bytes_on(listener, replies).await;
    let out = run(
        &sb,
        &url,
        &[
            "build",
            "run",
            "-o",
            "json",
            "--no-cache",
            "--",
            "cargo",
            "check",
        ],
    )
    .await;
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert_eq!(out.status.code(), Some(0), "{stderr}");
    let events: Vec<Value> = String::from_utf8_lossy(&out.stdout)
        .lines()
        .map(|line| serde_json::from_str(line).unwrap())
        .collect();
    assert_eq!(
        events
            .iter()
            .filter(|e| e["type"] == "queued" && e["retry"] == true)
            .count(),
        1
    );
    let running = events.iter().find(|e| e["type"] == "running").unwrap();
    assert_eq!(running["region"], retry_region);
    assert_eq!(running["volume"], false);
    assert_eq!(running["workspace"], "cold");
    assert_eq!(events.last().unwrap()["exit_code"], 0);
    let seen = log.lock();
    assert!(
        !seen.iter().any(|s| s.line.contains("/volumes")),
        "automatic retries must stay ephemeral"
    );
    let creates: Vec<Value> = seen
        .iter()
        .filter(|s| s.line.starts_with(&format!("POST /v1/{ENV}/leases ")))
        .map(|s| serde_json::from_str(&s.body).unwrap())
        .collect();
    assert_eq!(creates.len(), 2, "one initial lease and one retry");
    assert!(creates[0]["spec"]
        .get("region")
        .is_none_or(|r| r.as_str() == Some("")));
    assert_eq!(
        creates[1]["spec"]["region"], retry_region,
        "the retry must pin the accepted Cell"
    );
    for lease in &creates {
        assert!(
            lease["spec"]
                .get("volumes")
                .is_none_or(|v| v.as_array().is_some_and(|a| a.is_empty())),
            "{lease}"
        );
    }
    assert!(seen
        .iter()
        .any(|s| s.line.starts_with(&format!("GET /v1/{first} "))));
    let commands: Vec<_> = seen
        .iter()
        .filter(|s| s.body.contains("\"cargo\""))
        .collect();
    assert_eq!(
        commands.len(),
        1,
        "the lost attempt never started the command"
    );
    assert!(
        commands[0].body.contains(r#""/workspace",".","1","cargo""#),
        "cold lease-owned targets use scoped remote cache"
    );
    assert_eq!(
        seen.iter().filter(|s| s.line.contains(":release")).count(),
        1
    );
    assert!(seen
        .last()
        .unwrap()
        .line
        .contains(&format!("/{second}:release")));
}

/// An unavailable region refuses the volume-free lease immediately: 125,
/// not retryable, and no workspace request or allocation is attempted.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_region_not_offered_exits_125() {
    let sb = Sandbox::new("build-run-no-region");
    git_init(&sb.dir);
    std::fs::write(sb.dir.join("a.txt"), "a\n").unwrap();
    let no_cell = json!({"code": "SHAPE_NOT_OFFERED", "status": 422, "detail": "`spec.region`: region `gra` has no Sandboxes Cell", "retryable": false});
    let (url, log) = serve(vec![(422, no_cell)]).await;
    let out = run(
        &sb,
        &url,
        &[
            "build", "run", "-o", "json", "--region", "gra", "--", "true",
        ],
    )
    .await;
    assert_eq!(out.status.code(), Some(125));
    let stdout = String::from_utf8_lossy(&out.stdout);
    let last: Value = serde_json::from_str(stdout.lines().last().unwrap()).unwrap();
    assert_eq!(last["retryable"], false, "{last}");
    assert!(
        last["error"]
            .as_str()
            .unwrap()
            .contains("no Sandboxes Cell"),
        "{last}"
    );
    assert_volume_free_regional_leases(&log, "gra");
    assert_eq!(log.lock().len(), 1, "no retry, no wait, no Volume request");
}

/// Explicit regional placement executes cold without consulting Volume APIs.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_in_an_explicit_region_uses_a_cold_owned_workspace() {
    let sb = Sandbox::new("build-run-no-volumes");
    git_init(&sb.dir);
    std::fs::write(sb.dir.join("main.rs"), "fn main() {}\n").unwrap();
    let json_reply = |v: Value| (200u16, "application/json", v.to_string().into_bytes());
    let lease = format!("{ENV}/leases/l1");
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    drop(listener);
    let replies = vec![
        json_reply(
            json!({"name": lease, "status": {"state": "ready", "endpoints": {"guestUri": url, "e2bSandboxId": "ABCD1234"}}}),
        ),
        json_reply(json!({"token": "h.p.s", "generation": 1})),
        envd_process("", 0),
        json_reply(json!({})),
        json_reply(json!({})),
        envd_process("", 0),
        envd_process("", 0),
        envd_process("built in gra\n", 0),
        json_reply(json!({"name": lease, "status": {"state": "ending"}})),
    ];
    let (url2, log) = serve_bytes_at(&url, replies).await;
    assert_eq!(url, url2);
    let t = std::time::Instant::now();
    let out = run(
        &sb,
        &url,
        &[
            "build", "run", "-o", "json", "--region", "gra", "--", "cargo", "check",
        ],
    )
    .await;
    let err = String::from_utf8_lossy(&out.stderr);
    assert_eq!(out.status.code(), Some(0), "{err}");
    assert!(
        t.elapsed() < std::time::Duration::from_secs(30),
        "no queue-timeout is waited out: {:?}",
        t.elapsed()
    );
    let stdout = String::from_utf8_lossy(&out.stdout);
    let events: Vec<Value> = stdout
        .lines()
        .filter_map(|l| serde_json::from_str(l).ok())
        .collect();
    let running = events.iter().find(|e| e["type"] == "running").unwrap();
    assert_eq!(running["volume"], false, "{running}");
    assert_eq!(running["region"], "gra", "{running}");
    assert_eq!(running["workspace"], "cold", "{running}");
    let sync = events.iter().find(|e| e["type"] == "sync").unwrap();
    assert_eq!(sync["volume"], false, "{sync}");
    assert_eq!(sync["full"], true, "{sync}");
    let last = events.last().unwrap();
    assert_eq!(last["type"], "result");
    assert_eq!(last["exit_code"], 0, "{last}");

    assert_volume_free_regional_leases(&log, "gra");
    let seen = log.lock();
    let created: Value = serde_json::from_str(&seen[0].body).unwrap();
    assert_eq!(created["spec"]["shape"], "build-large", "{created}");
    let args = &seen[7].body;
    assert!(args.contains(r#""/workspace",".","1","cargo""#), "{args}");
    assert!(
        seen[8].line.contains(":release"),
        "the lease is always released"
    );
}

/// The toolchain cannot be installed: nothing the user asked for has started,
/// so the run is a platform failure (125), never the status of a command that
/// did not run, and the machine is released.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_exits_125_when_the_toolchain_cannot_be_installed() {
    let sb = Sandbox::new("provision-fails");
    project(&sb.dir);
    let (url, log) = build_stub_with(
        envd_process_err(
            "the Rust toolchain could not be installed:\nerror: could not download channel-rust-1.99.0.toml.sha256 (Connection timed out)\n",
            4,
        ),
        vec![ended()],
    )
    .await;
    let out = run(
        &sb,
        &url,
        &[
            "build",
            "run",
            "--region",
            "hk",
            "--no-cache",
            "--",
            "cargo",
            "build",
        ],
    )
    .await;
    let err = String::from_utf8_lossy(&out.stderr);
    assert_eq!(out.status.code(), Some(125), "{err}");
    assert!(err.contains("Connection timed out"), "{err}");
    assert!(err.contains("[retryable]"), "{err}");
    assert_volume_free_regional_leases(&log, "hk");
    let seen = log.lock();
    let lines: Vec<&str> = seen.iter().map(|s| s.line.as_str()).collect();
    assert_eq!(lines.len(), 8, "the command is never started: {lines:?}");
    assert!(lines[7].contains(":release"), "{lines:?}");
}

/// A process stream that breaks after the command started (a proxy on the way
/// restarted) is reattached to the same process: no second start, no second
/// machine, and the command's own exit code.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_reattaches_a_broken_stream_to_the_running_command() {
    let sb = Sandbox::new("stream-reattach");
    project(&sb.dir);
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let json_reply = |v: Value| (200u16, "application/json", v.to_string().into_bytes());
    let mut replies = attempt_replies(&url, envd_process("", 0));
    replies.push(envd_broken());
    replies.push(json_reply(json!({"token": "h.p.t", "generation": 1}))); // a fresh guest token
    replies.push(envd_process("after the break\n", 101)); // Connect by tag
    replies.push(ended());
    let (_, log) = serve_bytes_on(listener, replies).await;
    let out = run(
        &sb,
        &url,
        &[
            "build",
            "run",
            "--region",
            "hk",
            "--no-cache",
            "--",
            "cargo",
            "test",
        ],
    )
    .await;
    let err = String::from_utf8_lossy(&out.stderr);
    assert_eq!(out.status.code(), Some(101), "{err}");
    assert_eq!(String::from_utf8_lossy(&out.stdout), "after the break\n");
    assert!(err.contains("reattached to the running command"), "{err}");
    assert!(!err.contains("retrying"), "{err}");
    assert_volume_free_regional_leases(&log, "hk");
    let seen = log.lock();
    let start = seen
        .iter()
        .find(|s| s.body.contains("\"cargo\""))
        .expect("the command was started");
    assert!(
        start.line.contains("process.Process/Start"),
        "{}",
        start.line
    );
    let tag = start
        .body
        .split("\"tag\":\"")
        .nth(1)
        .unwrap()
        .split('"')
        .next()
        .unwrap()
        .to_string();
    assert!(tag.starts_with("sylphx-"), "{tag}");
    let connect = seen
        .iter()
        .find(|s| s.line.contains("process.Process/Connect"))
        .expect("the run reattached");
    assert!(
        connect
            .body
            .contains(&format!("{{\"process\":{{\"tag\":\"{tag}\"}}}}")),
        "{}",
        connect.body
    );
    assert_eq!(
        seen.iter().filter(|s| s.body.contains("\"cargo\"")).count(),
        1,
        "started once"
    );
    assert_eq!(
        seen.iter().filter(|s| s.line.contains(":release")).count(),
        1,
        "one machine"
    );
}

/// A machine lost after the command started ends the run with 137: the
/// command is never run again, and 125 stays for failures before it starts.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_exits_137_when_the_machine_is_lost_after_the_command_started() {
    let sb = Sandbox::new("stream-lost");
    project(&sb.dir);
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let json_reply = |v: Value| (200u16, "application/json", v.to_string().into_bytes());
    let mut replies = attempt_replies(&url, envd_process("", 0));
    replies.push(envd_broken());
    replies.push(json_reply(json!({"token": "h.p.t", "generation": 1})));
    // Connect: envd no longer knows the process.
    replies.push((
        200,
        "application/connect+json",
        envd_frame(
            2,
            json!({"error": {"code": "not_found", "message": "process not found"}}),
        ),
    ));
    replies.push(json_reply(
        json!({"name": format!("{ENV}/leases/l1"), "status": {"state": "ended", "endReason": "machine_lost"}}),
    ));
    replies.push(ended());
    let (_, log) = serve_bytes_on(listener, replies).await;
    let out = run(
        &sb,
        &url,
        &[
            "build",
            "run",
            "--region",
            "hk",
            "-o",
            "json",
            "--no-cache",
            "--",
            "cargo",
            "test",
        ],
    )
    .await;
    let err = String::from_utf8_lossy(&out.stderr);
    assert_eq!(out.status.code(), Some(137), "{err}");
    assert!(
        err.contains("machine_lost after the command started"),
        "{err}"
    );
    let result: Value = String::from_utf8_lossy(&out.stdout)
        .lines()
        .filter_map(|l| serde_json::from_str::<Value>(l).ok())
        .find(|v| v["type"] == "result")
        .expect("a result event");
    assert_eq!(result["outcome"], "lost");
    assert_eq!(result["exit_code"], 137);
    assert_eq!(result["retryable"], true);
    assert_volume_free_regional_leases(&log, "hk");
    let seen = log.lock();
    assert_eq!(
        seen.iter().filter(|s| s.body.contains("\"cargo\"")).count(),
        1,
        "never run again"
    );
}

/// Running out of space before command start retries on a fresh lease-owned
/// workspace, without reusing, creating, or emptying a persistent Volume.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_retries_a_full_workspace_once_on_a_fresh_lease() {
    let sb = Sandbox::new("ws-full-retry");
    project(&sb.dir);
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let mut replies: Vec<_> = attempt_replies(&url, envd_process("", 0))
        .into_iter()
        .take(2)
        .collect();
    replies.push(envd_process_err(
        "gzip: stdout: No space left on device\n",
        1,
    ));
    replies.push(ended()); // the full machine is released before the retry
    let mut retry = attempt_replies(&url, envd_process("", 0));
    let mut lease: Value = serde_json::from_slice(&retry[0].2).unwrap();
    lease["name"] = json!(format!("{ENV}/leases/l2"));
    retry[0].2 = lease.to_string().into_bytes();
    replies.extend(retry);
    replies.push(envd_process("built\n", 0));
    replies.push(ended());
    let (_, log) = serve_bytes_on(listener, replies).await;
    let out = run(
        &sb,
        &url,
        &[
            "build",
            "run",
            "--region",
            "hk",
            "--no-cache",
            "--",
            "cargo",
            "test",
        ],
    )
    .await;
    let err = String::from_utf8_lossy(&out.stderr);
    assert_eq!(out.status.code(), Some(0), "{err}");
    assert_eq!(String::from_utf8_lossy(&out.stdout), "built\n");
    assert!(err.contains("ran out of space; retrying once"), "{err}");
    assert_volume_free_regional_leases(&log, "hk");
    let seen = log.lock();
    let leases: Vec<Value> = seen
        .iter()
        .filter(|s| s.line.starts_with(&format!("POST /v1/{ENV}/leases ")))
        .map(|s| serde_json::from_str(&s.body).unwrap())
        .collect();
    assert_eq!(leases.len(), 2);
    let boots: Vec<_> = seen
        .iter()
        .filter(|s| s.body.contains("\"bootstrap\""))
        .collect();
    assert_eq!(boots.len(), 2);
    for boot in boots {
        assert!(
            boot.body
                .contains("\"bootstrap\",\"/workspace\",\"0\",\"0\""),
            "{}",
            boot.body
        );
    }
    assert_eq!(
        seen.iter().filter(|s| s.body.contains("\"cargo\"")).count(),
        1
    );
    assert_eq!(
        seen.iter().filter(|s| s.line.contains(":release")).count(),
        2
    );
}

/// A machine that stays queued past --queue-timeout: the lease is released
/// and the run answers 125, retryable, in about that long, not 30 minutes.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_queue_timeout_releases_and_exits_125() {
    queued_build_timeout("run").await;
}

#[tokio::test(flavor = "multi_thread")]
async fn build_image_queue_timeout_never_reserves_a_workspace_volume() {
    queued_build_timeout("image").await;
}

async fn queued_build_timeout(subcommand: &str) {
    let sb = Sandbox::new(&format!("build-{subcommand}-queue-timeout"));
    git_init(&sb.dir);
    std::fs::write(sb.dir.join("a.txt"), "a\n").unwrap();
    let lease = format!("{ENV}/leases/l1");
    let queued = json!({"name": lease, "spec": {"region": "hk"}, "status": {"state": "granted"}});
    let (url, log) = serve(vec![
        (200, queued.clone()),
        (200, queued.clone()),
        (200, queued.clone()),
        (200, queued.clone()),
        (200, queued),
    ])
    .await;
    let t = std::time::Instant::now();
    let mut args = vec![
        "build",
        subcommand,
        "-o",
        "json",
        "--region",
        "hk",
        "--queue-timeout",
        "3s",
    ];
    if subcommand == "run" {
        args.extend(["--fresh", "--", "true"]);
    }
    let out = run(&sb, &url, &args).await;
    assert!(
        t.elapsed() < std::time::Duration::from_secs(30),
        "{:?}",
        t.elapsed()
    );
    assert_eq!(out.status.code(), Some(125));
    let stdout = String::from_utf8_lossy(&out.stdout);
    let last: Value = serde_json::from_str(stdout.lines().last().unwrap()).unwrap();
    assert_eq!(last["retryable"], true, "{last}");
    assert!(
        last["error"].as_str().unwrap().contains("--queue-timeout"),
        "{last}"
    );
    assert_volume_free_regional_leases(&log, "hk");
    let seen = log.lock();
    assert!(
        seen.iter().all(|s| s.line.contains("/leases")),
        "queued work must not contact workspace or guest APIs"
    );
    assert!(
        seen.iter().any(|s| s.line.contains(":release")),
        "the queued lease is released"
    );
}

/// [`serve_bytes`] on a known address (the guest's URI must be known before
/// the replies are built).
async fn serve_bytes_at(url: &str, replies: Vec<(u16, &'static str, Vec<u8>)>) -> (String, Log) {
    let addr = url.trim_start_matches("http://").to_string();
    serve_bytes_on(TcpListener::bind(addr).await.unwrap(), replies).await
}

fn seat_fixture() -> Value {
    let w = |key: &str, secs: i64, util: f64| {
        json!({
            "window_key": key, "utilization": util, "used_percent": util * 100.0,
            "limit_window_seconds": secs, "reset_at": "2099-01-01T00:00:00.000Z",
            "observed_at": "2099-01-01T00:00:00.000Z", "limit_reached": util >= 1.0,
            "state": "AVAILABLE",
        })
    };
    json!({"object": "list", "data": [
        {"id": "seat-ok", "provider": "claude", "state": "active", "last_success_at": null,
         "quota_pressure": 0.4, "windows": [w("claude_5h", 18000, 0.1), w("claude_7d", 604800, 0.4)]},
        {"id": "seat-spent", "provider": "claude", "state": "active", "last_success_at": null,
         "quota_pressure": 1.0, "windows": [w("claude_5h", 18000, 1.0), w("claude_7d", 604800, 0.5)]},
        {"id": "seat-login", "provider": "claude", "state": "reauth_required", "last_success_at": null,
         "quota_pressure": null, "windows": []},
    ]})
}

#[tokio::test(flavor = "multi_thread")]
async fn ai_top_once_reads_the_operator_seats_and_shows_red_lines() {
    let (url, log) = serve(vec![(200, seat_fixture())]).await;
    let sb = Sandbox::new("ai-top");
    let out = run(&sb, &url, &["ai", "top", "--once"]).await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let text = String::from_utf8_lossy(&out.stdout);
    assert!(text.contains("1 usable / 1 spent / 1 out"), "{text}");
    assert!(
        text.contains("RED  seat seat-login (claude): login needed"),
        "{text}"
    );
    assert!(
        text.contains("SESSIONS AND SUBAGENTS") && text.contains("n/a"),
        "{text}"
    );
    assert!(!text.contains('\x1b'), "no colour off a terminal");
    let seen = log.lock();
    assert!(
        seen[0].line.starts_with("GET /v1/operator/seats "),
        "{}",
        seen[0].line
    );
    // the reading is kept for the pace
    let kept = std::fs::read_to_string(sb.dir.join("config/ai-top-history.json")).unwrap();
    assert!(kept.contains("seat-ok"), "{kept}");
}

#[tokio::test(flavor = "multi_thread")]
async fn ai_top_json_is_one_document_with_the_missing_fields_named() {
    let (url, _) = serve(vec![(200, seat_fixture())]).await;
    let sb = Sandbox::new("ai-top-json");
    let out = run(&sb, &url, &["ai", "top", "--json"]).await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let v: Value = serde_json::from_slice(&out.stdout).unwrap();
    assert_eq!(v["seats"]["usable"], 1);
    assert_eq!(v["seats"]["spent"], 1);
    assert!(v["sessions"].is_null());
    assert!(v["red"]
        .as_array()
        .unwrap()
        .iter()
        .any(|l| l.as_str().unwrap().contains("seat-login")));
    assert!(!v["missing_api_fields"].as_array().unwrap().is_empty());
}

#[tokio::test(flavor = "multi_thread")]
async fn ai_top_says_so_when_the_key_may_not_read_seats() {
    let (url, _) = serve(vec![(
        403,
        json!({"code": "missing_scope", "error": {"code": "missing_scope", "message": "needs ai:operator:seats:read"}}),
    )])
    .await;
    let sb = Sandbox::new("ai-top-403");
    let out = run(&sb, &url, &["ai", "top", "--once"]).await;
    assert_eq!(out.status.code(), Some(1));
    assert!(String::from_utf8_lossy(&out.stderr).contains("missing_scope"));
}

// ---- the shared build cache ------------------------------------------------

const CACHE_TOKEN: &str = "cache-token-0123456789-do-not-print";

fn cache_reply() -> Value {
    json!({
        "token": CACHE_TOKEN,
        "expires_at": "2026-10-04T12:00:00Z",
        "org": "org_a",
        "cache": "prj_a",
        "read_scopes": ["protected", "dev"],
        "write_scope": "dev",
        "env": {
            "SCCACHE_WEBDAV_ENDPOINT": "https://cache.test/sccache",
            "SCCACHE_WEBDAV_TOKEN": CACHE_TOKEN,
            "TURBO_API": "https://cache.test",
            "TURBO_TOKEN": CACHE_TOKEN,
            "TURBO_TEAM": "team_prj_a",
            "SCCACHE_IGNORE_SERVER_IO_ERROR": "1"
        }
    })
}

async fn run_with(
    sb: &Sandbox,
    url: &str,
    args: &[&str],
    tweak: impl FnOnce(&mut Command),
) -> std::process::Output {
    let mut c = sb.cmd(url, args);
    tweak(&mut c);
    tokio::task::spawn_blocking(move || c.output().unwrap())
        .await
        .unwrap()
}

/// A server that takes connections and never answers.
async fn hanging_server() -> String {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    tokio::spawn(async move {
        let mut held = Vec::new();
        while let Ok((sock, _)) = listener.accept().await {
            held.push(sock);
        }
    });
    url
}

/// The sandbox and guest stub up to command start: a volume-free ready lease,
/// its token, and a cold workspace sync.
/// `tail` is what happens from the command on.
async fn build_stub(tail: Vec<(u16, &'static str, Vec<u8>)>) -> (String, Log) {
    build_stub_with(envd_process("", 0), tail).await
}

/// One attempt: a volume-free ready lease, its token, a cold sync and provision.
fn attempt_replies(
    url: &str,
    provision: (u16, &'static str, Vec<u8>),
) -> Vec<(u16, &'static str, Vec<u8>)> {
    let json_reply = |v: Value| (200u16, "application/json", v.to_string().into_bytes());
    let lease = format!("{ENV}/leases/l1");
    vec![
        json_reply(
            json!({"name": lease, "spec": {"region": "hk"}, "status": {"state": "ready", "endpoints": {"guestUri": url, "e2bSandboxId": "ABCD1234"}}}),
        ),
        json_reply(json!({"token": "h.p.s", "generation": 1})),
        envd_process("", 0),
        json_reply(json!({})),
        json_reply(json!({})),
        envd_process("", 0),
        provision,
    ]
}

/// [`build_stub`] with the preparation step (the toolchain install) answering
/// `provision`.
async fn build_stub_with(
    provision: (u16, &'static str, Vec<u8>),
    tail: Vec<(u16, &'static str, Vec<u8>)>,
) -> (String, Log) {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let mut replies = attempt_replies(&url, provision);
    replies.extend(tail);
    serve_bytes_on(listener, replies).await
}

fn ended() -> (u16, &'static str, Vec<u8>) {
    (
        200,
        "application/json",
        json!({"name": format!("{ENV}/leases/l1"), "status": {"state": "ending"}})
            .to_string()
            .into_bytes(),
    )
}

/// The environment the command was started with (the eighth call).
fn started_env(log: &Log) -> Value {
    let seen = log.lock();
    let start = &seen[7];
    assert!(start.line.starts_with("POST /process.Process/Start"));
    let text = &start.body;
    let from = text.find("{\"process\"").expect("a JSON message");
    let end = text.rfind('}').unwrap();
    let v: Value = serde_json::from_str(&text[from..=end]).unwrap();
    v["process"]["envs"].clone()
}

/// Every regional Build fixture leases without any persistent workspace API.
fn assert_volume_free_regional_leases(log: &Log, region: &str) {
    let seen = log.lock();
    let mut leases = 0;
    for request in seen.iter() {
        assert!(!request.line.contains("/volumes"), "{}", request.line);
        if request.line.starts_with(&format!("POST /v1/{ENV}/leases ")) {
            let body: Value = serde_json::from_str(&request.body).unwrap();
            assert_eq!(body["spec"]["region"], region, "{body}");
            assert!(
                body["spec"]
                    .get("volumes")
                    .is_none_or(|v| v.as_array().is_some_and(|a| a.is_empty())),
                "{body}"
            );
            leases += 1;
        }
    }
    assert!(
        leases > 0,
        "the fixture must request a volume-free regional lease"
    );
}

fn sylphx_lines(stderr: &str) -> Vec<&str> {
    stderr
        .lines()
        .filter(|l| l.starts_with("sylphx:"))
        .collect()
}

fn project(dir: &std::path::Path) {
    git_init(dir);
    std::fs::write(dir.join("a.txt"), "a\n").unwrap();
}

#[tokio::test(flavor = "multi_thread")]
async fn build_run_mints_a_write_token_and_gives_the_command_the_cache_env() {
    let sb = Sandbox::new("cache-run");
    project(&sb.dir);
    let (cache, clog) = serve(vec![(200, cache_reply())]).await;
    let (url, log) = build_stub(vec![envd_process("built\n", 0), ended()]).await;
    let out = run_with(
        &sb,
        &url,
        &[
            "build",
            "run",
            "--region",
            "hk",
            "-o",
            "json",
            "--timeout",
            "10m",
            "--env",
            "TURBO_TEAM=mine",
            "--",
            "make",
        ],
        |c| {
            c.env("SYLPHX_BUILD_CACHE_URL", &cache);
        },
    )
    .await;
    let err = String::from_utf8_lossy(&out.stderr);
    let stdout = String::from_utf8_lossy(&out.stdout);
    assert_eq!(out.status.code(), Some(0), "{err}");
    assert_volume_free_regional_leases(&log, "hk");
    // The mint: the key as a bearer, and the run's access, network and life.
    let seen = clog.lock();
    assert_eq!(seen.len(), 1);
    assert!(
        seen[0].line.starts_with("POST /v1/tokens"),
        "{}",
        seen[0].line
    );
    assert!(
        seen[0]
            .head
            .to_ascii_lowercase()
            .contains("authorization: bearer sylphx_sk_test"),
        "{}",
        seen[0].head
    );
    let body: Value = serde_json::from_str(&seen[0].body).unwrap();
    assert_eq!(
        body,
        json!({"project": "prj_a", "access": "write", "ttl_seconds": 1500, "network": "cluster"})
    );
    // The command's environment: the cache's, the user's --env winning.
    let env = started_env(&log);
    assert_eq!(env["SCCACHE_WEBDAV_ENDPOINT"], "https://cache.test/sccache");
    assert_eq!(env["SCCACHE_WEBDAV_TOKEN"], CACHE_TOKEN);
    assert_eq!(env["TURBO_TEAM"], "mine");
    // The token is in no event, no log line and no error.
    assert!(!stdout.contains(CACHE_TOKEN), "{stdout}");
    assert!(!err.contains(CACHE_TOKEN), "{err}");
    let events: Vec<Value> = stdout
        .lines()
        .map(|l| serde_json::from_str(l).unwrap())
        .collect();
    let running = events.iter().find(|e| e["type"] == "running").unwrap();
    assert_eq!(running["cache"], true);
    assert!(!err.contains("build cache unavailable"), "{err}");
}

#[tokio::test(flavor = "multi_thread")]
async fn build_run_caps_the_token_life_at_the_longest_run() {
    let sb = Sandbox::new("cache-ttl");
    project(&sb.dir);
    let (cache, clog) = serve(vec![(200, cache_reply())]).await;
    let (url, log) = build_stub(vec![envd_process("", 0), ended()]).await;
    let out = run_with(
        &sb,
        &url,
        &[
            "build",
            "run",
            "--region",
            "hk",
            "--timeout",
            "6h",
            "--",
            "make",
        ],
        |c| {
            c.env("SYLPHX_BUILD_CACHE_URL", &cache);
        },
    )
    .await;
    assert_eq!(out.status.code(), Some(0));
    assert_volume_free_regional_leases(&log, "hk");
    let body: Value = serde_json::from_str(&clog.lock()[0].body).unwrap();
    assert_eq!(body["ttl_seconds"], 22500);
}

/// Whatever goes wrong minting, the run builds without the cache, says so on
/// exactly one line, and stays within three `sylphx:` lines.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_builds_without_the_cache_when_it_cannot_be_minted() {
    let hang = hanging_server().await;
    let (e503, _) = serve(vec![(
        503,
        json!({"error": {"code": "UNAVAILABLE", "message": "store down"}}),
    )])
    .await;
    let (e403, _) = serve(vec![(
        403,
        json!({"error": {"code": "PERMISSION_DENIED", "message": "no"}}),
    )])
    .await;
    let (e401, _) = serve(vec![(
        401,
        json!({"error": {"code": "UNAUTHENTICATED", "message": "no"}}),
    )])
    .await;
    let (e400, _) = serve(vec![(
        400,
        json!({"error": {"code": "INVALID_ARGUMENT", "message": "ttl"}}),
    )])
    .await;
    let (junk, _) = serve_bytes(vec![(200, "application/json", b"{not json".to_vec())]).await;
    let (noenv, _) = serve(vec![(200, json!({"token": CACHE_TOKEN}))]).await;
    let cases: Vec<(&str, String, &str)> = vec![
        ("503", e503, "HTTP 503 UNAVAILABLE"),
        ("403", e403, "HTTP 403 PERMISSION_DENIED"),
        ("401", e401, "HTTP 401 UNAUTHENTICATED"),
        ("400", e400, "HTTP 400 INVALID_ARGUMENT"),
        ("json", junk, "unreadable reply"),
        ("noenv", noenv, "unreadable reply"),
        ("down", "http://127.0.0.1:9".into(), "unreachable"),
        ("timeout", hang, "timed out"),
    ];
    for (name, cache, reason) in cases {
        let sb = Sandbox::new(&format!("cache-fail-{name}"));
        project(&sb.dir);
        let (url, log) = build_stub(vec![envd_process("ok\n", 0), ended()]).await;
        let out = run_with(
            &sb,
            &url,
            &["build", "run", "--region", "hk", "--", "make"],
            |c| {
                c.env("SYLPHX_BUILD_CACHE_URL", &cache);
            },
        )
        .await;
        let err = String::from_utf8_lossy(&out.stderr);
        assert_eq!(out.status.code(), Some(0), "{name}: {err}");
        assert_volume_free_regional_leases(&log, "hk");
        assert_eq!(String::from_utf8_lossy(&out.stdout), "ok\n", "{name}");
        let warn: Vec<&str> = err.lines().filter(|l| l.contains("build cache")).collect();
        assert_eq!(
            warn,
            [format!(
                "sylphx: warning: build cache unavailable ({reason}); building without it"
            )],
            "{name}: {err}"
        );
        assert!(sylphx_lines(&err).len() <= 3, "{name}: {err}");
        assert_eq!(started_env(&log), json!({}), "{name}");
        assert!(!err.contains(CACHE_TOKEN), "{name}");
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn build_run_no_cache_mints_nothing_and_warns_of_nothing() {
    let sb = Sandbox::new("cache-off");
    project(&sb.dir);
    let (cache, clog) = serve(vec![(200, cache_reply())]).await;
    let (url, log) = build_stub(vec![envd_process("", 0), ended()]).await;
    let out = run_with(
        &sb,
        &url,
        &["build", "run", "--region", "hk", "--no-cache", "--", "make"],
        |c| {
            c.env("SYLPHX_BUILD_CACHE_URL", &cache);
        },
    )
    .await;
    let err = String::from_utf8_lossy(&out.stderr);
    assert_eq!(out.status.code(), Some(0), "{err}");
    assert_volume_free_regional_leases(&log, "hk");
    assert!(clog.lock().is_empty(), "no token was asked for");
    assert!(!err.contains("build cache"), "{err}");
    assert_eq!(sylphx_lines(&err).len(), 3, "{err}");
    assert_eq!(started_env(&log), json!({}));
}

/// An error from the machine that happens to repeat the token does not
/// carry it out.
#[tokio::test(flavor = "multi_thread")]
async fn build_run_errors_never_carry_the_token() {
    let sb = Sandbox::new("cache-scrub");
    project(&sb.dir);
    let (cache, _) = serve(vec![(200, cache_reply())]).await;
    let echo = format!("boom {CACHE_TOKEN}").into_bytes();
    let (url, log) = build_stub(vec![
        (500, "text/plain", echo),
        (
            200,
            "application/json",
            json!({"name": format!("{ENV}/leases/l1"), "spec": {"region": "hk"}, "status": {"state": "ready"}})
                .to_string()
                .into_bytes(),
        ),
        ended(),
    ])
    .await;
    let out = run_with(
        &sb,
        &url,
        &["build", "run", "--region", "hk", "--", "make"],
        |c| {
            c.env("SYLPHX_BUILD_CACHE_URL", &cache);
        },
    )
    .await;
    let err = String::from_utf8_lossy(&out.stderr);
    assert_eq!(out.status.code(), Some(125), "{err}");
    assert_volume_free_regional_leases(&log, "hk");
    assert!(err.contains("sylphx: error:"), "{err}");
    assert!(!err.contains(CACHE_TOKEN), "{err}");
}

fn fake_sccache_dir(name: &str) -> PathBuf {
    let dir = std::env::temp_dir().join(format!("sylphx-sccache-{name}-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let bin = dir.join("sccache");
    std::fs::write(&bin, "#!/bin/sh\n").unwrap();
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap();
    }
    dir
}

#[tokio::test(flavor = "multi_thread")]
async fn build_cache_env_prints_shell_exports_from_a_read_token() {
    let sb = Sandbox::new("cache-env");
    let mut reply = cache_reply();
    reply["env"]["ODD"] = json!("it's $HOME `x` \"y\"");
    let (cache, clog) = serve(vec![(200, reply)]).await;
    let with = fake_sccache_dir("with");
    let out = run_with(&sb, "http://127.0.0.1:9", &["build", "cache", "env"], |c| {
        c.env("SYLPHX_BUILD_CACHE_URL", &cache)
            .env("PATH", &with)
            .env_remove("RUSTC_WRAPPER");
    })
    .await;
    let _ = std::fs::remove_dir_all(&with);
    let err = String::from_utf8_lossy(&out.stderr);
    assert_eq!(out.status.code(), Some(0), "{err}");
    let stdout = String::from_utf8_lossy(&out.stdout).to_string();
    assert!(
        stdout.contains(&format!("export SCCACHE_WEBDAV_TOKEN='{CACHE_TOKEN}'\n")),
        "{stdout}"
    );
    assert!(
        stdout.ends_with("export RUSTC_WRAPPER=sccache\n"),
        "{stdout}"
    );
    assert!(!err.contains(CACHE_TOKEN), "{err}");
    // A shell reads the quoting back to the same values.
    let script = format!("{stdout}\nprintf %s \"$ODD\"");
    let echoed = Command::new("sh").arg("-c").arg(script).output().unwrap();
    assert_eq!(
        String::from_utf8_lossy(&echoed.stdout),
        "it's $HOME `x` \"y\""
    );
    let seen = clog.lock();
    assert_eq!(seen.len(), 1);
    let body: Value = serde_json::from_str(&seen[0].body).unwrap();
    assert_eq!(
        body,
        json!({"project": "prj_a", "access": "read", "ttl_seconds": 43200, "network": "public"})
    );
    assert!(
        seen[0]
            .head
            .to_ascii_lowercase()
            .contains("authorization: bearer sylphx_sk_test"),
        "{}",
        seen[0].head
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn build_cache_env_adds_the_wrapper_only_when_it_is_missing_and_usable() {
    let with = fake_sccache_dir("wrapper");
    let without = std::env::temp_dir().join(format!("sylphx-no-sccache-{}", std::process::id()));
    std::fs::create_dir_all(&without).unwrap();
    for (name, path, wrapper, want) in [
        ("none", &without, None, false),
        ("set", &with, Some("other"), false),
        ("present", &with, None, true),
    ] {
        let sb = Sandbox::new(&format!("cache-wrapper-{name}"));
        let (cache, _) = serve(vec![(200, cache_reply())]).await;
        let out = run_with(&sb, "http://127.0.0.1:9", &["build", "cache", "env"], |c| {
            c.env("SYLPHX_BUILD_CACHE_URL", &cache).env("PATH", path);
            match wrapper {
                Some(w) => c.env("RUSTC_WRAPPER", w),
                None => c.env_remove("RUSTC_WRAPPER"),
            };
        })
        .await;
        assert_eq!(out.status.code(), Some(0), "{name}");
        let stdout = String::from_utf8_lossy(&out.stdout);
        assert_eq!(stdout.contains("RUSTC_WRAPPER"), want, "{name}: {stdout}");
    }
    let _ = std::fs::remove_dir_all(&with);
    let _ = std::fs::remove_dir_all(&without);
}

#[tokio::test(flavor = "multi_thread")]
async fn build_cache_env_json_is_the_env_object_and_takes_a_project() {
    let sb = Sandbox::new("cache-env-json");
    let (cache, clog) = serve(vec![(200, cache_reply())]).await;
    let out = run_with(
        &sb,
        "http://127.0.0.1:9",
        &[
            "build",
            "cache",
            "env",
            "--project",
            "prj_other",
            "-o",
            "json",
        ],
        |c| {
            c.env("SYLPHX_BUILD_CACHE_URL", &cache);
        },
    )
    .await;
    assert_eq!(out.status.code(), Some(0));
    let v: Value = serde_json::from_slice(&out.stdout).unwrap();
    assert_eq!(v, cache_reply()["env"]);
    let body: Value = serde_json::from_str(&clog.lock()[0].body).unwrap();
    assert_eq!(body["project"], "prj_other");
}

#[tokio::test(flavor = "multi_thread")]
async fn build_cache_env_failures_exit_125_or_2_and_say_why() {
    let (e503, _) = serve(vec![(
        503,
        json!({"error": {"code": "UNAVAILABLE", "message": "store down"}}),
    )])
    .await;
    let (e403, _) = serve(vec![(
        403,
        json!({"error": {"code": "PERMISSION_DENIED", "message": "key lacks build:cache.read"}}),
    )])
    .await;
    let (e400, _) = serve(vec![(
        400,
        json!({"error": {"code": "INVALID_ARGUMENT", "message": "unknown project"}}),
    )])
    .await;
    let (junk, _) = serve_bytes(vec![(200, "application/json", b"<html>".to_vec())]).await;
    let hang = hanging_server().await;
    for (name, cache, code, said) in [
        ("503", e503, 125, "store down"),
        ("403", e403, 125, "key lacks build:cache.read"),
        ("400", e400, 2, "unknown project"),
        ("junk", junk, 125, "unreadable reply"),
        ("down", "http://127.0.0.1:9".to_string(), 125, "unreachable"),
        ("hang", hang, 125, "timed out"),
        ("http", "http://cache.example.com".to_string(), 2, "https"),
    ] {
        let sb = Sandbox::new(&format!("cache-env-fail-{name}"));
        let out = run_with(&sb, "http://127.0.0.1:9", &["build", "cache", "env"], |c| {
            c.env("SYLPHX_BUILD_CACHE_URL", &cache);
        })
        .await;
        let err = String::from_utf8_lossy(&out.stderr);
        assert_eq!(out.status.code(), Some(code), "{name}: {err}");
        assert!(
            err.contains("error:") && err.contains(said),
            "{name}: {err}"
        );
        assert!(out.stdout.is_empty(), "{name}: nothing on stdout");
        assert!(!err.contains("sylphx_sk_test"), "{name}: {err}");
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn build_cache_env_needs_a_project_and_a_key() {
    // No linked project and a key that names none: usage, and no mint.
    let sb = Sandbox::new("cache-env-noproject");
    std::fs::remove_file(sb.dir.join(".sylphx/project.json")).unwrap();
    let (api, _) = serve(vec![(200, json!({"org": "orgs/org_a"}))]).await;
    let (cache, clog) = serve(vec![(200, cache_reply())]).await;
    let out = run_with(&sb, &api, &["build", "cache", "env"], |c| {
        c.env("SYLPHX_BUILD_CACHE_URL", &cache);
    })
    .await;
    assert_eq!(out.status.code(), Some(2));
    assert!(String::from_utf8_lossy(&out.stderr).contains("no project"));
    assert!(clog.lock().is_empty());
    // Signed out: a platform failure, as for `build run`.
    let out = run_with(&sb, "http://127.0.0.1:9", &["build", "cache", "env"], |c| {
        c.env_remove("SYLPHX_API_KEY");
    })
    .await;
    assert_eq!(out.status.code(), Some(125));
    assert!(String::from_utf8_lossy(&out.stderr).contains("not signed in"));
}

/// The test vector secret of the Standard Webhooks reference libraries; not a
/// credential.
const LISTEN_SECRET: &str = "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw"; // gitleaks:allow

/// `sylphx events listen --forward URL`: a temporary Queue subscribed to the
/// env's bus, each leased event POSTed to the local receiver with a valid
/// Standard Webhooks signature, acked, and the Queue and Subscription deleted
/// on the way out.
#[tokio::test(flavor = "multi_thread")]
async fn events_listen_relays_a_published_event_signed_to_the_local_url() {
    use base64::Engine as _;
    let (url, log) = serve(vec![
        (200, json!({"name": format!("{ENV}/queues/q")})),
        (200, json!({"name": format!("{ENV}/subscriptions/s")})),
        (
            200,
            json!({"messages": [{
                "message_id": "01HMSG",
                "event": {"id": "evt-1", "source": "test", "type": "order.created",
                          "data": {"n": 1}},
                "lease": {"message_id": "01HMSG", "lease_generation": "3"},
                "delivery_count": 1
            }]}),
        ),
        (200, json!({})),
        (200, json!({})),
        (200, json!({})),
    ])
    .await;
    let (hook, received) = serve(vec![(204, json!({}))]).await;
    let sb = Sandbox::new("events-listen");
    let forward = format!("{hook}/hook");
    let out = run(
        &sb,
        &url,
        &[
            "events",
            "listen",
            "--forward",
            &forward,
            "--secret",
            LISTEN_SECRET,
            "--count",
            "1",
            "-o",
            "json",
        ],
    )
    .await;
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );

    // The receiver got the CloudEvent, signed under the secret.
    let got = received.lock()[0].clone();
    assert!(got.line.starts_with("POST /hook "), "{}", got.line);
    let header = |name: &str| {
        got.head
            .lines()
            .find_map(|l| {
                let (k, v) = l.split_once(':')?;
                k.eq_ignore_ascii_case(name).then(|| v.trim().to_string())
            })
            .unwrap_or_default()
    };
    assert_eq!(header("webhook-id"), "01HMSG");
    let timestamp = header("webhook-timestamp");
    let now = now_secs() as i64;
    assert!((now - timestamp.parse::<i64>().unwrap()).abs() < 300);
    let key = base64::engine::general_purpose::STANDARD
        .decode(LISTEN_SECRET.trim_start_matches("whsec_"))
        .unwrap();
    let mut mac = <hmac::Hmac<sha2::Sha256> as hmac::Mac>::new_from_slice(&key).unwrap();
    hmac::Mac::update(
        &mut mac,
        format!("01HMSG.{timestamp}.{}", got.body).as_bytes(),
    );
    let expected = format!(
        "v1,{}",
        base64::engine::general_purpose::STANDARD.encode(hmac::Mac::finalize(mac).into_bytes())
    );
    assert_eq!(header("webhook-signature"), expected);
    assert_eq!(
        serde_json::from_str::<Value>(&got.body).unwrap(),
        json!({"specversion": "1.0", "id": "evt-1", "source": "test",
               "type": "order.created", "data": {"n": 1}})
    );

    // The listener's own calls, in order.
    let log = log.lock().clone();
    let lines: Vec<&str> = log.iter().map(|s| s.line.as_str()).collect();
    assert!(
        lines[0].starts_with(&format!("POST /v1/{ENV}/queues?queue_id=listen-")),
        "{lines:?}"
    );
    assert!(
        lines[1].starts_with(&format!(
            "POST /v1/{ENV}/subscriptions?subscription_id=listen-"
        )),
        "{lines:?}"
    );
    let sub: Value = serde_json::from_str(&log[1].body).unwrap();
    assert_eq!(sub["spec"]["topic"], format!("{ENV}/topics/default"));
    assert!(sub["spec"]["queue"]
        .as_str()
        .unwrap()
        .starts_with(&format!("{ENV}/queues/listen-")));
    assert!(lines[2].contains(":lease "), "{lines:?}");
    assert!(lines[3].contains(":ack "), "{lines:?}");
    let ack: Value = serde_json::from_str(&log[3].body).unwrap();
    assert_eq!(ack["leases"][0]["message_id"], "01HMSG");
    assert!(
        lines[4].starts_with(&format!("DELETE /v1/{ENV}/subscriptions/listen-")),
        "{lines:?}"
    );
    assert!(
        lines[5].starts_with(&format!("DELETE /v1/{ENV}/queues/listen-")),
        "{lines:?}"
    );

    let printed: Value = serde_json::from_slice(&out.stdout).unwrap();
    assert_eq!(printed["status"], 204);
    assert_eq!(printed["type"], "order.created");
}