//! End-to-end host behavior with a raw WASI guest.
use sausage::{Host, world::World};
use serde_json::{Value, json};
use std::{path::PathBuf, sync::OnceLock};
fn fixture() -> PathBuf {
    static WASM: OnceLock<PathBuf> = OnceLock::new();
    WASM.get_or_init(|| {
        let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .parent()
            .unwrap()
            .to_path_buf();
        let target = root.join("target/core/guests");
        let status = std::process::Command::new("cargo")
            .args([
                "build",
                "--offline",
                "--target",
                "wasm32-wasip2",
                "--manifest-path",
            ])
            .arg(root.join("cli/tests/plugins/probe/Cargo.toml"))
            .arg("--target-dir")
            .arg(&target)
            .status()
            .unwrap();
        assert!(status.success());
        target.join("wasm32-wasip2/debug/probe.wasm")
    })
    .clone()
}
struct TestWorld(PathBuf);
impl TestWorld {
    fn new() -> Self {
        let p = std::env::temp_dir().join(format!("sausage-{}", uuid::Uuid::now_v7()));
        let w = World::create(&p).unwrap();
        std::fs::create_dir_all(w.plugin_dir("probe")).unwrap();
        std::fs::copy(fixture(), w.plugin_dir("probe").join("plugin.wasm")).unwrap();
        Self(p)
    }
    async fn start(&self) -> Host {
        Host::start(&self.0).await.unwrap()
    }
}
impl Drop for TestWorld {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.0);
    }
}
async fn call(h: &Host, m: &str, p: Value) -> Value {
    tokio::time::timeout(std::time::Duration::from_secs(20), h.call(m, p))
        .await
        .unwrap()
        .unwrap()
}

/// PROTOCOL 3.1: json rpc lines carry results.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn json_rpc_lines_carry_results() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(call(&h, "probe.echo", json!({"x":1})).await, json!({"x":1}));

    h.stop().await;
}

/// PROTOCOL 3.2: method names split at the first dot.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn method_names_split_at_the_first_dot() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(
        call(&h, "probe.rpc.discover", json!({"x":2})).await,
        json!({"x":2})
    );

    h.stop().await;
}

/// PROTOCOL 3.3: processes can call other plugins.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn processes_can_call_other_plugins() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(
        call(
            &h,
            "probe.call",
            json!({"method":"probe.echo","params":{"nested":true}})
        )
        .await,
        json!({"nested":true})
    );

    h.stop().await;
}

/// PROTOCOL 3.3: pending ids distinguish numbers from strings.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn pending_ids_distinguish_numbers_from_strings() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(
        call(&h, "probe.multicall", json!({})).await,
        json!({"count":2})
    );

    h.stop().await;
}

/// PROTOCOL 3.3: notifications start attached processes.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn notifications_start_attached_processes() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(
        &h,
        "probe.notification",
        json!({"path":"/state/notification","text":"delivered"}),
    )
    .await;
    assert_eq!(
        std::fs::read_to_string(w.0.join("state/notification")).unwrap(),
        "delivered"
    );

    h.stop().await;
}

/// PROTOCOL 3.3: unknown response ids are dropped.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unknown_response_ids_are_dropped() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(
        h.call("probe.unknownid", json!({})).await.unwrap_err()["code"],
        -32002
    );

    h.stop().await;
}

/// PROTOCOL 3.8: process environment identifies the caller.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn process_environment_identifies_the_caller() {
    let w = TestWorld::new();
    let h = w.start().await;
    let env = call(&h, "probe.env", json!({})).await;
    assert_eq!(env["plugin"], "probe");
    assert_eq!(env["parent"], "console");
    assert!(env["pid"].as_str().unwrap().starts_with("probe-"));

    h.stop().await;
}

/// PROTOCOL 3.6: processes read and write the world.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn processes_read_and_write_the_world() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(
        &h,
        "probe.write",
        json!({"path":"/state/test","text":"hello"}),
    )
    .await;
    assert_eq!(
        call(&h, "probe.read", json!({"path":"/state/test"})).await,
        "hello"
    );

    h.stop().await;
}

/// PROTOCOL 3.9: missing plugins return minus 32001.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn missing_plugins_return_minus_32001() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(
        h.call("absent.run", json!({})).await.unwrap_err()["code"],
        -32001
    );

    h.stop().await;
}

/// PROTOCOL 9: unimplemented core methods return method not found.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unimplemented_core_methods_return_method_not_found() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(
        h.call("core.unknown", json!({})).await.unwrap_err()["code"],
        -32601
    );

    h.stop().await;
}

/// PROTOCOL 3.1: params must be objects.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn params_must_be_objects() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(
        h.call("probe.echo", json!([])).await.unwrap_err()["code"],
        -32602
    );

    h.stop().await;
}

/// PROTOCOL 3.9: exit without answer reports status and last logs.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exit_without_answer_reports_status_and_last_logs() {
    let w = TestWorld::new();
    let h = w.start().await;
    let e = h.call("probe.exit", json!({})).await.unwrap_err();
    assert_eq!(e["code"], -32002);
    assert!(
        e["data"]["lines"]
            .as_array()
            .unwrap()
            .iter()
            .any(|l| l["line"] == "last line")
    );

    h.stop().await;
}

/// PROTOCOL 3.5: progress notifications reach the caller.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn progress_notifications_reach_the_caller() {
    let w = TestWorld::new();
    let h = w.start().await;
    let mut notifications = h.subscribe();
    call(&h, "probe.notify", json!({})).await;
    assert_eq!(
        notifications.recv().await.unwrap()["params"]["name"],
        "progress"
    );

    h.stop().await;
}

/// PROTOCOL 3.5: every stdout notification reaches the caller before the following result.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stdout_notifications_precede_the_result() {
    let w = TestWorld::new();
    let h = w.start().await;
    let mut notifications = h.subscribe();
    for _ in 0..10 {
        assert_eq!(call(&h, "probe.burst", json!({"count":64})).await, 64);
        let mut call_id = None;
        for index in 0..64 {
            // A delayed notification fails here, even if it arrives later.
            let notification = notifications.try_recv().unwrap();
            assert_eq!(notification["params"]["name"], "burst");
            assert_eq!(notification["params"]["params"]["index"], index);
            if let Some(id) = &call_id {
                assert_eq!(&notification["params"]["call"], id);
            } else {
                call_id = Some(notification["params"]["call"].clone());
            }
        }
        assert!(matches!(
            notifications.try_recv(),
            Err(tokio::sync::broadcast::error::TryRecvError::Empty)
        ));
    }
    h.stop().await;
}

/// PROTOCOL 3.1: non message stdout warns once without its content.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn non_message_stdout_warns_once_without_its_content() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(&h, "probe.junk", json!({})).await;
    tokio::time::timeout(std::time::Duration::from_secs(5), async {
        while !h.processes().is_empty() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    let starts = call(&h, "core.log", json!({"grep":"start junk"})).await;
    let pid = &starts[0]["pid"];
    let warnings = call(&h, "core.log", json!({"pid":pid,"grep":"discarded stdout"})).await;
    assert_eq!(warnings.as_array().unwrap().len(), 1);
    assert!(!warnings.to_string().contains("junk"));

    h.stop().await;
}

/// PROTOCOL 3.1: stderr lines enter the journal.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stderr_lines_enter_the_journal() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(&h, "probe.junk", json!({})).await;
    h.stop().await;
    let log = call(&h, "core.log", json!({"grep":"stderr line"})).await;
    assert_eq!(log.as_array().unwrap().len(), 1);

    h.stop().await;
}

/// PROTOCOL 6: try lock reports held without waiting.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn try_lock_reports_held_without_waiting() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(
        call(&h, "core.lock", json!({"path":"x"})).await,
        json!({"held":true})
    );
    assert_eq!(
        call(&h, "core.lock", json!({"path":"x","wait":false})).await,
        json!({"held":false})
    );
    call(&h, "core.unlock", json!({"path":"x"})).await;
    assert_eq!(
        call(&h, "probe.lock", json!({"path":"x"})).await,
        json!({"held":true})
    );

    h.stop().await;
}

/// PROTOCOL 9: kill ends a process at once.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn kill_ends_a_process_at_once() {
    let w = TestWorld::new();
    let h = w.start().await;
    let pid = call(&h, "probe.serve", json!({})).await;
    call(&h, "core.kill", json!({"pid":pid})).await;
    assert!(!h.processes().contains_key(pid.as_str().unwrap()));

    h.stop().await;
}

/// PROTOCOL 9: native calls have process ids and results.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn native_calls_have_process_ids_and_results() {
    let w = TestWorld::new();
    let h = w.start().await;
    h.register_native(
        "native",
        std::sync::Arc::new(|ctx, method, params| {
            Box::pin(async move {
                ctx.notify("native", json!({}));
                Ok(json!({"pid":ctx.pid,"method":method,"params":params}))
            })
        }),
    );
    let result = call(&h, "native.echo", json!({"x":1})).await;
    assert_eq!(result["method"], "echo");
    assert!(result["pid"].as_str().unwrap().starts_with("native-"));

    h.stop().await;
}

/// PROTOCOL 2: reserved world roots prevent startup.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reserved_world_roots_prevent_startup() {
    let w = TestWorld::new();
    let h = w.start().await;
    for name in ["proc", "run", "self"] {
        std::fs::create_dir(w.0.join(name)).unwrap();
        assert!(World::open(&w.0).is_err());
        std::fs::remove_dir(w.0.join(name)).unwrap();
    }
    h.stop().await;
}

/// PROTOCOL 8: boot waits for each readiness result before the next service.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn boot_waits_for_readiness_in_order() {
    let w = TestWorld::new();
    std::fs::write(
        w.0.join("config/boot.json"),
        serde_json::to_vec(&json!([
            {"method":"probe.write","params":{"path":"/state/boot","text":"ready"}},
            {"method":"probe.read","params":{"path":"/state/boot"}}
        ]))
        .unwrap(),
    )
    .unwrap();
    let h = w.start().await;
    assert_eq!(
        std::fs::read_to_string(w.0.join("state/boot")).unwrap(),
        "ready"
    );

    h.stop().await;
}

/// PROTOCOL 3.6: attached children stop when the parent ends.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn attached_children_stop_when_the_parent_ends() {
    let w = TestWorld::new();
    let h = w.start().await;
    let attached = call(
        &h,
        "probe.call",
        json!({"method":"probe.serve","params":{}}),
    )
    .await;
    tokio::time::timeout(std::time::Duration::from_secs(7), async {
        while h.processes().contains_key(attached.as_str().unwrap()) {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();

    h.stop().await;
}

/// PROTOCOL 3.4: detached calls have no parent and outlive the caller.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn detached_calls_have_no_parent_and_outlive_the_caller() {
    let w = TestWorld::new();
    let h = w.start().await;
    let detached = call(
        &h,
        "probe.call",
        json!({"method":"core.call","params":{"method":"probe.serve","params":{},"detach":true}}),
    )
    .await;
    assert!(h.processes()[detached.as_str().unwrap()].detached);
    call(&h, "core.stop", json!({"pid":detached})).await;

    h.stop().await;
}

/// PROTOCOL 3.8: busy guests do not block other calls.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn busy_guests_do_not_block_other_calls() {
    let w = TestWorld::new();
    let h = w.start().await;
    let hc = h.clone();
    let busy = tokio::spawn(async move { hc.call("probe.busy", json!({})).await });
    let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async {
        loop {
            if let Some((pid, _)) = h.processes().into_iter().find(|(_, p)| p.method == "busy") {
                break pid;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    assert_eq!(
        call(&h, "probe.echo", json!({"works":true})).await,
        json!({"works":true})
    );
    call(&h, "core.kill", json!({"pid":pid})).await;
    assert_eq!(busy.await.unwrap().unwrap_err()["code"], -32002);

    h.stop().await;
}

/// PROTOCOL 6: same process lock acquisitions queue.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn same_process_lock_acquisitions_queue() {
    let w = TestWorld::new();
    let h = w.start().await;

    h.register_native(
        "locks",
        std::sync::Arc::new(|ctx, _, _| {
            Box::pin(async move {
                assert_eq!(
                    ctx.call("core.lock", json!({"path":"fifo"})).await.unwrap(),
                    json!({"held":true})
                );
                assert_eq!(
                    ctx.call("core.lock", json!({"path":"fifo","wait":false}))
                        .await
                        .unwrap(),
                    json!({"held":false})
                );
                let mut second = Box::pin(ctx.call("core.lock", json!({"path":"fifo"})));
                std::future::poll_fn(|cx| {
                    assert!(second.as_mut().poll(cx).is_pending());
                    std::task::Poll::Ready(())
                })
                .await;
                ctx.call("core.unlock", json!({"path":"fifo"}))
                    .await
                    .unwrap();
                assert_eq!(second.await.unwrap(), json!({"held":true}));
                Ok(json!({}))
            })
        }),
    );
    call(&h, "locks.test", json!({})).await;
    h.stop().await;
    assert_eq!(
        call(&h, "core.lock", json!({"path":"fifo","wait":false})).await,
        json!({"held":true})
    );
    h.stop().await;
}

/// PROTOCOL 3.8: tcp udp and dns.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn tcp_udp_and_dns() {
    let w = TestWorld::new();
    let h = w.start().await;

    use tokio::io::AsyncWriteExt;
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let tcp = format!("localhost:{}", listener.local_addr().unwrap().port());
    let udp = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
    let udp_addr = udp.local_addr().unwrap().to_string();
    let server = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        stream.write_all(b"tcp").await.unwrap();
    });
    assert_eq!(
        call(&h, "probe.network", json!({"tcp":tcp,"udp":udp_addr})).await,
        "tcp"
    );
    let mut buf = [0; 3];
    udp.recv_from(&mut buf).await.unwrap();
    assert_eq!(&buf, b"udp");
    server.await.unwrap();

    h.stop().await;
}

/// PROTOCOL 6: native kill drops future and releases locks.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn native_kill_drops_future_and_releases_locks() {
    let w = TestWorld::new();
    let h = w.start().await;

    use std::sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    };
    struct Dropped(Arc<AtomicBool>);
    impl Drop for Dropped {
        fn drop(&mut self) {
            self.0.store(true, Ordering::SeqCst);
        }
    }
    let dropped = Arc::new(AtomicBool::new(false));
    let flag = dropped.clone();
    h.register_native(
        "native",
        Arc::new(move |ctx, _, _| {
            let flag = flag.clone();
            Box::pin(async move {
                let _guard = Dropped(flag);
                ctx.call("core.lock", json!({"path":"native-lock"}))
                    .await
                    .unwrap();
                ctx.notify("ready", json!({"pid":ctx.pid}));
                std::future::pending().await
            })
        }),
    );
    let mut notifications = h.subscribe();
    let hc = h.clone();
    let pending = tokio::spawn(async move { hc.call("native.wait", json!({})).await });
    let notification =
        tokio::time::timeout(std::time::Duration::from_secs(10), notifications.recv())
            .await
            .unwrap()
            .unwrap();
    let pid = &notification["params"]["params"]["pid"];
    assert_eq!(
        call(&h, "core.lock", json!({"path":"native-lock","wait":false})).await,
        json!({"held":false})
    );
    call(&h, "core.kill", json!({"pid":pid})).await;
    tokio::time::timeout(std::time::Duration::from_secs(1), async {
        while !dropped.load(Ordering::SeqCst) {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    assert_eq!(
        call(&h, "core.lock", json!({"path":"native-lock","wait":false})).await,
        json!({"held":true})
    );
    assert_eq!(pending.await.unwrap().unwrap_err()["code"], -32002);

    h.stop().await;
}

/// PROTOCOL 3.6: shutdown stops busy processes in parallel after five seconds.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn shutdown_stops_busy_processes_in_parallel_after_five_seconds() {
    let w = TestWorld::new();
    let h = w.start().await;

    let mut tasks = Vec::new();
    for _ in 0..2 {
        let hc = h.clone();
        tasks.push(tokio::spawn(async move {
            hc.call("probe.busy", json!({})).await
        }));
    }
    tokio::time::timeout(std::time::Duration::from_secs(20), async {
        while h.processes().len() != 2 {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    let before = std::time::Instant::now();
    h.stop().await;
    assert!(before.elapsed() >= std::time::Duration::from_millis(4900));
    assert!(before.elapsed() < std::time::Duration::from_secs(7));
    for task in tasks {
        assert_eq!(task.await.unwrap().unwrap_err()["code"], -32002);
    }
    assert!(h.processes().is_empty());
    h.stop().await;
}

/// PROTOCOL 3.6: stopping rejects calls but allows notifications.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stopping_rejects_calls_but_allows_notifications() {
    let w = TestWorld::new();
    let h = w.start().await;

    use std::sync::{Arc, Mutex};
    let (tx, rx) = tokio::sync::oneshot::channel();
    let tx = Arc::new(Mutex::new(Some(tx)));
    let gate = Arc::new(tokio::sync::Notify::new());
    let release = gate.clone();
    h.register_native(
        "native",
        Arc::new(move |ctx, _, _| {
            let tx = tx.lock().unwrap().take().unwrap();
            let gate = gate.clone();
            Box::pin(async move {
                tx.send(ctx.clone()).ok();
                gate.notified().await;
                Ok(json!({}))
            })
        }),
    );
    let hc = h.clone();
    let pending = tokio::spawn(async move { hc.call("native.wait", json!({})).await });
    let ctx = rx.await.unwrap();
    let hc = h.clone();
    let pid = ctx.pid.clone();
    let stop = tokio::spawn(async move { hc.call("core.stop", json!({"pid":pid})).await });
    tokio::time::timeout(std::time::Duration::from_secs(2), async {
        while h.processes()[&ctx.pid].state != "stopping" {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    assert_eq!(
        ctx.call("probe.echo", json!({})).await.unwrap_err()["code"],
        -32600
    );
    let mut notifications = h.subscribe();
    ctx.call("core.notify", json!({"name":"stop-hook","params":{}}))
        .await
        .unwrap();
    assert_eq!(
        notifications.recv().await.unwrap()["params"]["name"],
        "stop-hook"
    );
    release.notify_one();
    assert_eq!(pending.await.unwrap().unwrap(), json!({}));
    assert_eq!(stop.await.unwrap().unwrap(), json!({}));

    h.stop().await;
}

/// PROTOCOL 9: stop closes stdin and waits for exit.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stop_closes_stdin_and_waits_for_exit() {
    let w = TestWorld::new();
    let h = w.start().await;
    let hc = h.clone();
    let pending = tokio::spawn(async move { hc.call("probe.wait_eof", json!({})).await });
    let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async {
        loop {
            if let Some((pid, _)) = h
                .processes()
                .into_iter()
                .find(|(_, p)| p.method == "wait_eof")
            {
                break pid;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    call(&h, "core.stop", json!({"pid":pid})).await;
    assert_eq!(
        std::fs::read_to_string(w.0.join("state/stopped")).unwrap(),
        "yes"
    );
    assert_eq!(pending.await.unwrap().unwrap_err()["code"], -32002);

    h.stop().await;
}

/// PROTOCOL 9: core exit requests restart from its handler.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn core_exit_requests_restart_from_its_handler() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(call(&h, "core.exit", json!({})).await, json!({}));
    tokio::time::timeout(std::time::Duration::from_secs(1), h.exit_requested())
        .await
        .unwrap();

    h.stop().await;
}

/// PROTOCOL 3.3: process ids are uuidv7 and versions are sha256.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn process_ids_are_uuidv7_and_versions_are_sha256() {
    let w = TestWorld::new();
    let h = w.start().await;
    use sha2::{Digest, Sha256};
    let pid = call(&h, "probe.serve", json!({"version":1})).await;
    let pid = pid.as_str().unwrap();
    let info = h.processes()[pid].clone();
    assert_eq!(
        info.version,
        format!("{:x}", Sha256::digest(std::fs::read(fixture()).unwrap()))
    );
    assert_eq!(info.plugin, "probe");
    assert_eq!(info.method, "serve");
    assert_eq!(info.params, json!({"version":1}));
    assert_eq!(info.parent.as_deref(), Some("console"));
    assert_eq!(info.state, "running");
    assert!(!info.detached);
    assert!(info.start_time > 0);
    let uuid = uuid::Uuid::parse_str(pid.strip_prefix("probe-").unwrap()).unwrap();
    assert_eq!(uuid.get_version_num(), 7);
    call(&h, "core.stop", json!({"pid":pid})).await;

    h.stop().await;
}

/// PROTOCOL 7: journal filters and entries persist in sqlite.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn journal_filters_and_entries_persist_in_sqlite() {
    let w = TestWorld::new();
    let h = w.start().await;
    let pid = call(&h, "probe.serve", json!({})).await;
    call(&h, "core.stop", json!({"pid":pid})).await;
    let entries = call(
        &h,
        "core.log",
        json!({"plugin":"probe","pid":pid,"limit":2}),
    )
    .await;
    assert_eq!(entries.as_array().unwrap().len(), 2);
    let entry = &entries[0];
    let id = &entry["id"];
    let single = call(&h, "core.log", json!({"since":id,"until":id})).await;
    assert_eq!(single, json!([entry]));
    h.stop().await;
    drop(h);
    let h = w.start().await;
    assert_eq!(
        call(&h, "core.log", json!({"since":id,"until":id})).await,
        single
    );
    h.stop().await;
    assert!(
        std::fs::read(w.0.join("state/core/journal.sqlite"))
            .unwrap()
            .starts_with(b"SQLite format 3")
    );
    h.stop().await;
}

/// PROTOCOL 3.1: blank stdout lines do not warn.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn blank_stdout_lines_do_not_warn() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(&h, "probe.blank", json!({})).await;
    h.stop().await;
    let warnings = call(&h, "core.log", json!({"grep":"discarded stdout"})).await;
    assert_eq!(warnings, json!([]));
    h.stop().await;
}

/// PROTOCOL 3.1: large requests and responses cross pipe capacity.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn large_requests_and_responses_cross_pipe_capacity() {
    let w = TestWorld::new();
    let h = w.start().await;
    let params = json!({"text":"x".repeat(192*1024)});
    assert_eq!(call(&h, "probe.echo", params.clone()).await, params);
    let outgoing = json!({"method":"probe.echo","params":params});
    assert_eq!(call(&h, "probe.call", outgoing).await, params);
    h.stop().await;
}

/// PROTOCOL 3.8: two plugins execute calls at the same time.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn two_plugins_execute_calls_at_the_same_time() {
    let w = TestWorld::new();
    let h = w.start().await;
    let other = w.0.join("plugins/other");
    std::fs::create_dir_all(&other).unwrap();
    std::fs::copy(fixture(), other.join("plugin.wasm")).unwrap();
    let hc = h.clone();
    let slow = tokio::spawn(async move { hc.call("probe.sleep", json!({"ms":1500})).await });
    while !h.processes().values().any(|p| p.method == "sleep") {
        tokio::task::yield_now().await;
    }
    assert_eq!(
        call(&h, "other.echo", json!({"ready":true})).await,
        json!({"ready":true})
    );
    assert!(!slow.is_finished());
    slow.await.unwrap().unwrap();
    h.stop().await;
}

/// PROTOCOL 3.3: request id is the process id.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn request_id_is_the_process_id() {
    let w = TestWorld::new();
    let h = w.start().await;
    let pid = call(&h, "probe.request_id", json!({})).await;
    let uuid =
        uuid::Uuid::parse_str(pid.as_str().unwrap().strip_prefix("probe-").unwrap()).unwrap();
    assert_eq!(uuid.get_version_num(), 7);
    h.stop().await;
}

/// PROTOCOL 9: core call returns the target result.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn core_call_returns_the_target_result() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(
        call(
            &h,
            "core.call",
            json!({"method":"probe.echo","params":{"n":1}})
        )
        .await,
        json!({"n":1})
    );
    h.stop().await;
}

/// PROTOCOL 9: core unlock releases the named lock.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn core_unlock_releases_the_named_lock() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(&h, "core.lock", json!({"path":"unlock"})).await;
    call(&h, "core.unlock", json!({"path":"unlock"})).await;
    assert_eq!(
        call(&h, "core.lock", json!({"path":"unlock","wait":false})).await,
        json!({"held":true})
    );
    h.stop().await;
}

/// PROTOCOL 6: locks do not block file writes.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn locks_do_not_block_file_writes() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(&h, "core.lock", json!({"path":"/state/advisory"})).await;
    call(
        &h,
        "probe.write",
        json!({"path":"/state/advisory","text":"allowed"}),
    )
    .await;
    assert_eq!(
        std::fs::read_to_string(w.0.join("state/advisory")).unwrap(),
        "allowed"
    );
    h.stop().await;
}

/// PROTOCOL 6: crashes release all process locks.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn crashes_release_all_process_locks() {
    let w = TestWorld::new();
    let h = w.start().await;
    h.register_native(
        "crash",
        std::sync::Arc::new(|ctx, _, _| {
            Box::pin(async move {
                ctx.call("core.lock", json!({"path":"crash"}))
                    .await
                    .unwrap();
                panic!("test crash");
            })
        }),
    );
    assert_eq!(
        h.call("crash.run", json!({})).await.unwrap_err()["code"],
        -32002
    );
    assert_eq!(
        call(&h, "core.lock", json!({"path":"crash","wait":false})).await,
        json!({"held":true})
    );
    h.stop().await;
}

/// PROTOCOL 7: journal entries have uuid time identity stream and line.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn journal_entries_have_uuid_time_identity_stream_and_line() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(&h, "probe.junk", json!({})).await;
    h.stop().await;
    let entries = call(&h, "core.log", json!({"plugin":"probe"})).await;
    for entry in entries.as_array().unwrap() {
        assert_eq!(
            uuid::Uuid::parse_str(entry["id"].as_str().unwrap())
                .unwrap()
                .get_version_num(),
            7
        );
        assert!(entry["time"].as_str().unwrap().contains('T'));
        assert!(entry["time"].as_str().unwrap().ends_with('Z'));
        assert!(entry["pid"].as_str().unwrap().starts_with("probe-"));
        assert!(["stderr", "core"].contains(&entry["stream"].as_str().unwrap()));
        assert!(entry["line"].is_string());
    }
    h.stop().await;
}

/// PROTOCOL 7: journal has wal and identity indexes.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn journal_has_wal_and_identity_indexes() {
    let w = TestWorld::new();
    let h = w.start().await;
    h.stop().await;
    let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
    assert_eq!(
        db.query_row("PRAGMA journal_mode", [], |r| r.get::<_, String>(0))
            .unwrap(),
        "wal"
    );
    for name in ["journal_pid", "journal_plugin"] {
        assert_eq!(
            db.query_row(
                "SELECT count(*) FROM sqlite_master WHERE type='index' AND name=?1",
                [name],
                |r| r.get::<_, i64>(0)
            )
            .unwrap(),
            1
        );
    }
    h.stop().await;
}

/// PROTOCOL 9: core exit works through core call.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn core_exit_works_through_core_call() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(
        call(&h, "core.call", json!({"method":"core.exit","params":{}})).await,
        json!({})
    );
    assert!(h.restart_requested());
    h.stop().await;
}

/// PROTOCOL 6: acquisitions from different processes keep their FIFO positions.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn lock_acquisitions_from_different_processes_are_fifo() {
    let w = TestWorld::new();
    let h = w.start().await;
    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
    let gate = std::sync::Arc::new(tokio::sync::Semaphore::new(0));
    let release = gate.clone();
    h.register_native(
        "queued",
        std::sync::Arc::new(move |ctx, _, _| {
            let tx = tx.clone();
            let gate = gate.clone();
            Box::pin(async move {
                tx.send(ctx).unwrap();
                gate.acquire().await.unwrap().forget();
                Ok(json!({}))
            })
        }),
    );
    let first_host = h.clone();
    let first = tokio::spawn(async move { first_host.call("queued.wait", json!({})).await });
    let second_host = h.clone();
    let second = tokio::spawn(async move { second_host.call("queued.wait", json!({})).await });
    let first_ctx = rx.recv().await.unwrap();
    let second_ctx = rx.recv().await.unwrap();
    call(&h, "core.lock", json!({"path":"ordered"})).await;
    let mut first_lock = Box::pin(first_ctx.call("core.lock", json!({"path":"ordered"})));
    let mut second_lock = Box::pin(second_ctx.call("core.lock", json!({"path":"ordered"})));
    std::future::poll_fn(|cx| {
        assert!(first_lock.as_mut().poll(cx).is_pending());
        assert!(second_lock.as_mut().poll(cx).is_pending());
        std::task::Poll::Ready(())
    })
    .await;
    call(&h, "core.unlock", json!({"path":"ordered"})).await;
    assert_eq!(first_lock.await.unwrap(), json!({"held":true}));
    std::future::poll_fn(|cx| {
        assert!(second_lock.as_mut().poll(cx).is_pending());
        std::task::Poll::Ready(())
    })
    .await;
    first_ctx
        .call("core.unlock", json!({"path":"ordered"}))
        .await
        .unwrap();
    assert_eq!(second_lock.await.unwrap(), json!({"held":true}));
    release.add_permits(2);
    first.await.unwrap().unwrap();
    second.await.unwrap().unwrap();
    h.stop().await;
}

/// PROTOCOL 3.5: notifications after the result do not reach the caller.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn notifications_after_the_result_are_dropped() {
    let w = TestWorld::new();
    let h = w.start().await;
    let (tx, rx) = tokio::sync::oneshot::channel();
    let tx = std::sync::Mutex::new(Some(tx));
    h.register_native(
        "native",
        std::sync::Arc::new(move |ctx, _, _| {
            tx.lock().unwrap().take().unwrap().send(ctx).ok();
            Box::pin(async { Ok(json!({})) })
        }),
    );
    let mut notifications = h.subscribe();
    call(&h, "native.answer", json!({})).await;
    let ctx = rx.await.unwrap();
    ctx.notify("late", json!({}));
    assert!(matches!(
        notifications.try_recv(),
        Err(tokio::sync::broadcast::error::TryRecvError::Empty)
    ));
    h.stop().await;
}

/// PROTOCOL 3.1: non-UTF-8 stdout is discarded, not treated as blank framing.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn non_utf8_stdout_is_not_blank_framing() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(&h, "probe.invalid_utf8", json!({})).await;
    h.stop().await;
    let warnings = call(&h, "core.log", json!({"grep":"discarded stdout"})).await;
    assert_eq!(warnings.as_array().unwrap().len(), 1);
}

/// PROTOCOL 8: a failed service enters the journal and does not prevent later services.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn failed_boot_services_are_logged_and_boot_continues() {
    let w = TestWorld::new();
    let services = json!([
        {"method":"absent.start","params":{}},
        {"method":"probe.write","params":{"path":"/state/continued","text":"yes"}}
    ]);
    std::fs::write(
        w.0.join("config/boot.json"),
        serde_json::to_vec(&services).unwrap(),
    )
    .unwrap();
    let h = w.start().await;
    assert_eq!(
        std::fs::read_to_string(w.0.join("state/continued")).unwrap(),
        "yes"
    );
    let errors = call(
        &h,
        "core.log",
        json!({"plugin":"core","grep":"no such plugin"}),
    )
    .await;
    assert_eq!(errors.as_array().unwrap().len(), 1);
    h.stop().await;
}

/// PROTOCOL 8: boot services are detached and remain alive after their readiness result.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn boot_services_are_detached() {
    let w = TestWorld::new();
    std::fs::write(
        w.0.join("config/boot.json"),
        br#"[{"method":"probe.serve","params":{}}]"#,
    )
    .unwrap();
    let h = w.start().await;
    let processes = h.processes();
    let services: Vec<_> = processes.values().filter(|p| p.method == "serve").collect();
    assert_eq!(services.len(), 1);
    assert!(services[0].detached);
    h.stop().await;
}

/// PROTOCOL 7: the core records start parameters, stop, kill, and exit status.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn core_lifetime_events_enter_the_journal() {
    let w = TestWorld::new();
    let h = w.start().await;
    let stopped = call(&h, "probe.serve", json!({"n":1})).await;
    call(&h, "core.stop", json!({"pid":stopped})).await;
    let killed = call(&h, "probe.serve", json!({"n":2})).await;
    call(&h, "core.kill", json!({"pid":killed})).await;
    let entries = call(&h, "core.log", json!({"plugin":"probe","limit":100})).await;
    let lines: Vec<_> = entries
        .as_array()
        .unwrap()
        .iter()
        .map(|entry| entry["line"].as_str().unwrap())
        .collect();
    assert!(lines.contains(&"start serve {\"n\":1}"));
    for line in ["stop", "kill", "exit 0", "killed"] {
        assert!(lines.contains(&line), "missing {line}");
    }
    h.stop().await;
}

/// PROTOCOL 4: the root merges world files and mounts, and cross-mount rename fails.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn filesystem_root_and_mount_boundaries() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(&h, "probe.filesystem", json!({"topic":"root"})).await;
    assert_eq!(
        std::fs::read_to_string(w.0.join("state/root")).unwrap(),
        "world"
    );
    h.stop().await;
}

/// PROTOCOL 4: a process publishes self files read-only under its own live proc folder.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn filesystem_self_and_proc_lifetime() {
    use sausage_vfs::Layer;
    let w = TestWorld::new();
    let h = w.start().await;
    let view = h.filesystem().await.unwrap();
    let pid = call(&h, "probe.filesystem", json!({"topic":"self"})).await;
    call(&h, "core.stop", json!({"pid":pid})).await;
    assert!(
        view.stat(&format!("/proc/{}", pid.as_str().unwrap()), true)
            .await
            .is_err()
    );
    let live = call(&h, "probe.self_serve", json!({})).await;
    call(&h, "probe.filesystem", json!({"topic":"isolated"})).await;
    assert_eq!(
        call(
            &h,
            "probe.read",
            json!({"path":format!("/proc/{}/endpoint",live.as_str().unwrap())})
        )
        .await,
        "endpoint"
    );
    call(&h, "core.stop", json!({"pid":live})).await;
    assert!(
        view.stat(&format!("/proc/{}", live.as_str().unwrap()), true)
            .await
            .is_err()
    );
    h.stop().await;
}

/// PROTOCOL 4: core-owned self names reject creation and replacement.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn filesystem_reserved_self_names() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(&h, "probe.filesystem", json!({"topic":"reserved"})).await;
    h.stop().await;
}

/// PROTOCOL 4: generated logs capture a finite snapshot at open.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn filesystem_log_snapshot_ends() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(&h, "probe.filesystem", json!({"topic":"snapshot"})).await;
    h.stop().await;
}

/// PROTOCOL 4: proc/core provides protocol, graph, lock, and plugin inspection files.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn filesystem_core_inspection_files() {
    let w = TestWorld::new();
    let h = w.start().await;
    call(&h, "probe.filesystem", json!({"topic":"core"})).await;
    h.stop().await;
}

/// PROTOCOL 4: proc/core exposes each discovered document and its generated Rust client.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn filesystem_discovery_document_and_client() {
    use sausage_vfs::{Layer, Open};
    let w = TestWorld::new();
    let document =
        json!({"methods":[{"name":"echo","params":[],"result":{"schema":{"type":"string"}}}]});
    std::fs::create_dir_all(w.0.join("state/probe")).unwrap();
    std::fs::write(w.0.join("state/probe/discover.json"), document.to_string()).unwrap();
    let h = w.start().await;
    let view = h.filesystem().await.unwrap();
    tokio::time::timeout(std::time::Duration::from_secs(20), async {
        loop {
            if view
                .open("/proc/core/plugins/probe/discover.json", Open::default())
                .await
                .is_ok()
            {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    let published = call(
        &h,
        "probe.read",
        json!({"path":"/proc/core/plugins/probe/discover.json"}),
    )
    .await;
    assert_eq!(
        serde_json::from_str::<Value>(published.as_str().unwrap()).unwrap(),
        document
    );
    let client = call(
        &h,
        "probe.read",
        json!({"path":"/proc/core/plugins/probe/client.rs"}),
    )
    .await;
    assert_eq!(
        client,
        sausage_interface::client_rs("probe", &document).unwrap()
    );
    h.stop().await;
}

/// PROTOCOL 4: run is shared across processes but empty after a host restart.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn filesystem_run_is_shared_and_ephemeral() {
    use sausage_vfs::Layer;
    let w = TestWorld::new();
    let h = w.start().await;
    call(
        &h,
        "probe.write",
        json!({"path":"/run/shared","text":"yes"}),
    )
    .await;
    assert_eq!(
        call(&h, "probe.read", json!({"path":"/run/shared"})).await,
        "yes"
    );
    h.stop().await;
    let h = w.start().await;
    assert!(
        h.filesystem()
            .await
            .unwrap()
            .list("/run")
            .await
            .unwrap()
            .is_empty()
    );
    h.stop().await;
}

/// PROTOCOL 6: no write from a killed process lands after the next holder acquires its lock.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn filesystem_kill_drains_writes_before_lock_transfer() {
    let w = TestWorld::new();
    let h = w.start().await;
    let pid = call(&h, "probe.write_loop", json!({})).await;
    let next = h.clone();
    let waiter = tokio::spawn(async move { call(&next, "probe.locked_write", json!({})).await });
    use sausage_vfs::{Layer, Open};
    let view = h.filesystem().await.unwrap();
    tokio::time::timeout(std::time::Duration::from_secs(20), async {
        loop {
            let file = view
                .open("/proc/core/locks.json", Open::default())
                .await
                .unwrap();
            let locks: Value = serde_json::from_slice(&file.read(0, 65536).await.unwrap()).unwrap();
            if locks["/state/loop"]["waiting"]
                .as_array()
                .is_some_and(|queue| !queue.is_empty())
            {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    call(&h, "core.kill", json!({"pid":pid})).await;
    waiter.await.unwrap();
    assert_eq!(
        std::fs::read_to_string(w.0.join("state/loop")).unwrap(),
        "new"
    );
    h.stop().await;
}

/// PROTOCOL 4: native processes appear in proc and locks.json identifies holders and waiters.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn filesystem_native_processes_and_lock_snapshot() {
    use sausage_vfs::{Layer, Open};
    let w = TestWorld::new();
    let h = w.start().await;
    h.register_native(
        "native",
        std::sync::Arc::new(|ctx, _, _| {
            Box::pin(async move {
                ctx.call("core.lock", json!({"path":"live"})).await.unwrap();
                ctx.notify("ready", json!({"pid":ctx.pid}));
                std::future::pending().await
            })
        }),
    );
    let mut events = h.subscribe();
    let host = h.clone();
    let task = tokio::spawn(async move { host.call("native.wait", json!({})).await });
    let event = events.recv().await.unwrap();
    let pid = event["params"]["params"]["pid"].as_str().unwrap();
    let view = h.filesystem().await.unwrap();
    assert!(
        view.list("/proc")
            .await
            .unwrap()
            .iter()
            .any(|e| e.name == pid)
    );
    let info = call(
        &h,
        "probe.read",
        json!({"path":format!("/proc/{pid}/info.json")}),
    )
    .await;
    assert_eq!(
        serde_json::from_str::<Value>(info.as_str().unwrap()).unwrap()["version"],
        "native"
    );
    let host = h.clone();
    let waiter = tokio::spawn(async move { host.call("core.lock", json!({"path":"live"})).await });
    tokio::time::timeout(std::time::Duration::from_secs(5), async {
        loop {
            let file = view
                .open("/proc/core/locks.json", Open::default())
                .await
                .unwrap();
            let locks: Value = serde_json::from_slice(&file.read(0, 65536).await.unwrap()).unwrap();
            if locks["live"]["waiting"] == json!(["console"]) {
                assert_eq!(locks["live"]["owner"], pid);
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    call(&h, "core.kill", json!({"pid":pid})).await;
    assert_eq!(waiter.await.unwrap().unwrap(), json!({"held":true}));
    assert!(task.await.unwrap().is_err());
    assert!(view.stat(&format!("/proc/{pid}"), true).await.is_err());
    h.stop().await;
}