//! End-to-end host behavior with a raw WASI guest.
use super::common::probe as fixture;
use sausage::{Host, world::World};
use serde_json::{Value, json};
use std::path::PathBuf;
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_with_cache(
            &self.0,
            std::path::Path::new(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache")),
        )
        .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.3: a closed stdin does not discard a response already on stdout.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn closed_stdin_drains_stdout() {
    let w = TestWorld::new();
    std::fs::write(w.0.join("state/early"), "answer").unwrap();
    let h = w.start().await;
    assert_eq!(
        call(&h, "probe.echo", json!({"large":"x".repeat(2*1024*1024)})).await,
        "early"
    );
    h.stop().await;
}

/// PROTOCOL 3.3: failed imports remain process failures, even with initial input in flight.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn failed_imports_preserve_process_status() {
    let w = TestWorld::new();
    let path = w.0.join("plugins/probe/plugin.wasm");
    let mut bytes = std::fs::read(&path).unwrap();
    for index in 0..bytes.len() - 5 {
        if &bytes[index..index + 5] == b"wasi:" {
            bytes[index..index + 5].copy_from_slice(b"fake:");
        }
    }
    std::fs::write(path, bytes).unwrap();
    let h = w.start().await;
    let error = tokio::time::timeout(
        std::time::Duration::from_secs(20),
        h.call("probe.echo", json!({"large":"x".repeat(2*1024*1024)})),
    )
    .await
    .unwrap()
    .unwrap_err();
    assert_eq!(error["code"], -32002);
    assert!(error["data"]["status"].as_str().unwrap().contains("fake:"));
    h.stop().await;
}

/// PROTOCOL 3.3: a closed stdin preserves the actual exit status and final stderr.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn closed_stdin_preserves_failure_details() {
    let w = TestWorld::new();
    std::fs::write(w.0.join("state/early"), "fail").unwrap();
    let h = w.start().await;
    let error = tokio::time::timeout(
        std::time::Duration::from_secs(20),
        h.call("probe.echo", json!({"large":"x".repeat(2*1024*1024)})),
    )
    .await
    .unwrap()
    .unwrap_err();
    assert_eq!(error["code"], -32002);
    // WASIp2 command.run represents a nonzero exit as its unit error result.
    assert_eq!(error["data"]["status"], "exit 1");
    assert!(
        error["data"]["lines"]
            .as_array()
            .unwrap()
            .iter()
            .any(|line| line["line"] == "last early log")
    );
    h.stop().await;
}

/// PROTOCOL 7: invalid stderr bytes do not stop later log lines or the guest.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stderr_uses_lossy_utf8_and_continues() {
    let w = TestWorld::new();
    let h = w.start().await;
    let pid = call(&h, "probe.invalid_stderr", json!({})).await;
    call(&h, "core.stop", json!({"pid":pid})).await;
    let logs = call(&h, "core.log", json!({"pid":pid})).await;
    assert!(
        logs.as_array()
            .unwrap()
            .iter()
            .any(|line| line["line"] == "\u{fffd}")
    );
    assert!(
        logs.as_array()
            .unwrap()
            .iter()
            .any(|line| line["line"] == "after invalid UTF-8")
    );
    h.stop().await;
}

/// PROTOCOL 3.3 and 7: stderr drains even while a large initial stdin write blocks.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stderr_flood_does_not_block_stdin() {
    let w = TestWorld::new();
    std::fs::write(w.0.join("state/early"), "flood").unwrap();
    let h = w.start().await;
    let payload = json!({"large":"x".repeat(1024*1024)});
    assert_eq!(call(&h, "probe.echo", payload.clone()).await, payload);
    h.stop().await;
    let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
    let count: i64 = db
        .query_row(
            "SELECT count(*) FROM journal WHERE line LIKE 'flood %'",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert!(count >= 5000);
}

/// PROTOCOL 3.6: shutdown rejects new processes, including calls after stop returns.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn shutdown_rejects_new_processes() {
    let w = TestWorld::new();
    let h = w.start().await;
    h.stop().await;
    assert_eq!(
        h.call("probe.echo", json!({})).await.unwrap_err()["code"],
        -32600
    );
    assert!(h.processes().is_empty());
}

/// PROTOCOL 6: stdout unlock takes effect before the next result reaches its caller.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stdout_unlock_precedes_result() {
    let w = TestWorld::new();
    let h = w.start().await;
    for _ in 0..10 {
        let pid = call(&h, "probe.unlock_answer", json!({})).await;
        assert_eq!(
            call(
                &h,
                "core.lock",
                json!({"path":"ordered-unlock","wait":false})
            )
            .await,
            json!({"held":true})
        );
        call(&h, "core.unlock", json!({"path":"ordered-unlock"})).await;
        call(&h, "core.stop", json!({"pid":pid})).await;
    }
    h.stop().await;
}

/// PROTOCOL 3.5: a notification-started process has no call to receive progress.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn notification_process_has_no_progress_destination() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(call(&h, "probe.notification_child", json!({})).await, 0);
    h.stop().await;
}

/// PROTOCOL 3.1: an explicit null error does not replace a successful result.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn null_error_is_success() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(call(&h, "probe.null_error", json!({})).await, 42);
    h.stop().await;
}

/// PROTOCOL 7: a journal insertion failure does not poison the host or stop later writes.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn journal_recovers_after_sqlite_write_failure() {
    let w = TestWorld::new();
    let h = w.start().await;
    let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
    db.busy_timeout(std::time::Duration::from_secs(5)).unwrap();
    db.execute_batch("CREATE TRIGGER reject_writes BEFORE INSERT ON journal BEGIN SELECT RAISE(FAIL,'test failure'); END;").unwrap();
    assert_eq!(
        call(&h, "probe.echo", json!({"failed_log":true})).await,
        json!({"failed_log":true})
    );
    call(&h, "core.log", json!({})).await;
    db.execute_batch("DROP TRIGGER reject_writes;").unwrap();
    assert_eq!(
        call(&h, "probe.echo", json!({"recovered":true})).await,
        json!({"recovered":true})
    );
    let entries = call(&h, "core.log", json!({"grep":"recovered"})).await;
    assert!(!entries.as_array().unwrap().is_empty());
    h.stop().await;
}

/// PROTOCOL 4: trailing slashes preserve live proc folders and generated plugin folders.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn proc_directories_accept_trailing_slashes() {
    use sausage_vfs::{Layer, Open};
    let w = TestWorld::new();
    let document = json!({"methods":[]});
    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.list("/proc/core/plugins/probe").await.is_ok() {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    let pid = call(&h, "probe.serve", json!({})).await;
    for path in [
        "/proc/".to_string(),
        "/proc/core/".into(),
        "/proc/core/plugins/".into(),
        "/proc/core/plugins/probe/".into(),
        format!("/proc/{}/", pid.as_str().unwrap()),
    ] {
        assert!(view.stat(&path, true).await.unwrap().directory);
        view.list(&path).await.unwrap();
        view.open(
            &path,
            Open {
                directory: true,
                ..Open::default()
            },
        )
        .await
        .unwrap();
    }
    h.stop().await;
}

/// PROTOCOL 3.4: Unix replacements with the same size and mtime use the new component.
#[cfg(unix)]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn component_identity_includes_inode_and_ctime() {
    let w = TestWorld::new();
    let path = w.0.join("plugins/probe/plugin.wasm");
    let mut bytes = std::fs::read(&path).unwrap();
    bytes.extend_from_slice(&[0, 2, 1, b'a']);
    std::fs::write(&path, &bytes).unwrap();
    let modified = std::fs::metadata(&path).unwrap().modified().unwrap();
    let h = w.start().await;
    let old = call(&h, "probe.serve", json!({})).await;
    *bytes.last_mut().unwrap() = b'b';
    std::fs::write(&path, &bytes).unwrap();
    std::fs::File::options()
        .write(true)
        .open(&path)
        .unwrap()
        .set_times(std::fs::FileTimes::new().set_modified(modified))
        .unwrap();
    let new = call(&h, "probe.serve", json!({})).await;
    let table = h.processes();
    assert_ne!(
        table[old.as_str().unwrap()].version,
        table[new.as_str().unwrap()].version
    );
    h.stop().await;
}

/// PROTOCOL 3.6 and 6: stopping cancels a native process's queued lock request immediately.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stopping_removes_queued_lock_acquisitions() {
    use sausage_vfs::{Layer, Open};
    let w = TestWorld::new();
    let h = w.start().await;
    call(&h, "core.lock", json!({"path":"queued-stop"})).await;
    h.register_native(
        "queued",
        std::sync::Arc::new(|ctx, _, _| {
            Box::pin(async move {
                ctx.notify("ready", json!({"pid":ctx.pid}));
                ctx.call("core.lock", json!({"path":"queued-stop"})).await
            })
        }),
    );
    let mut events = h.subscribe();
    let host = h.clone();
    let pending = tokio::spawn(async move { host.call("queued.wait", json!({})).await });
    let event = events.recv().await.unwrap();
    let pid = event["params"]["params"]["pid"].clone();
    let view = h.filesystem().await.unwrap();
    tokio::time::timeout(std::time::Duration::from_secs(10), async {
        loop {
            let f = view
                .open("/proc/core/locks.json", Open::default())
                .await
                .unwrap();
            let locks: Value = serde_json::from_slice(&f.read(0, 65536).await.unwrap()).unwrap();
            if locks["queued-stop"]["waiting"] == json!([pid]) {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    tokio::time::timeout(
        std::time::Duration::from_secs(1),
        h.call("core.stop", json!({"pid":pid})),
    )
    .await
    .unwrap()
    .unwrap();
    assert!(pending.await.unwrap().is_err());
    call(&h, "core.unlock", json!({"path":"queued-stop"})).await;
    assert_eq!(
        call(&h, "core.lock", json!({"path":"queued-stop","wait":false})).await,
        json!({"held":true})
    );
    h.stop().await;
}

/// 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.yaml"),
        serde_yaml_ng::to_string(&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: host shutdown stops busy processes together after five seconds.
/// PROTOCOL 3.6: core.stop terminates an attached tree within the same five-second bound.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stop_deadlines() {
    tokio::join!(
        shutdown_stops_busy_processes_in_parallel_after_five_seconds(),
        attached_tree_stops_in_parallel(),
    );
}

/// Checks the shutdown half of the shared deadline test.
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()
            .values()
            .filter(|p| p.plugin == "probe" && p.method == "busy")
            .count()
            != 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 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.wait_eof", json!({})).await });
    let pid = tokio::time::timeout(std::time::Duration::from_secs(10), async {
        while !w.0.join("state/waiting").exists() {
            tokio::task::yield_now().await;
        }
        loop {
            if let Some((pid, _)) = h
                .processes()
                .into_iter()
                .find(|(_, p)| p.method == "wait_eof")
            {
                break pid;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    assert_eq!(
        call(&h, "other.echo", json!({"ready":true})).await,
        json!({"ready":true})
    );
    assert!(!slow.is_finished());
    call(&h, "core.stop", json!({"pid":pid})).await;
    assert_eq!(slow.await.unwrap().unwrap_err()["code"], -32002);
    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 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.yaml"),
        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.yaml"),
        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.tree", json!({"depth":0})).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;
}

/// PROTOCOL 2: invalid plugin names cannot route calls, and core cannot be shadowed.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn plugin_names_and_core_reservation() {
    let w = TestWorld::new();
    let h = w.start().await;
    for name in ["Upper", "bad!", "with space", "é", "", "../probe"] {
        h.register_native(
            name,
            std::sync::Arc::new(|_, _, _| Box::pin(async { Ok(json!("bad")) })),
        );
        assert_eq!(
            h.call(&format!("{name}.echo"), json!({}))
                .await
                .unwrap_err()["code"],
            -32601
        );
    }
    h.register_native(
        "core",
        std::sync::Arc::new(|_, _, _| Box::pin(async { Ok(json!("bad")) })),
    );
    assert_eq!(
        h.call("core.unknown", json!({})).await.unwrap_err()["code"],
        -32601
    );
    for name in ["a-z_09", "0"] {
        h.register_native(
            name,
            std::sync::Arc::new(|_, _, _| Box::pin(async { Ok(json!(true)) })),
        );
        assert_eq!(call(&h, &format!("{name}.echo"), json!({})).await, true);
    }
    h.stop().await;
}

/// PROTOCOL 3.1: a batch on guest stdout does not execute any of its requests.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn batches_are_discarded() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(call(&h, "probe.batch", json!({})).await, true);
    h.stop().await;
    assert!(!w.0.join("state/batch").exists());
    assert_eq!(
        call(&h, "core.log", json!({"grep":"discarded stdout"}))
            .await
            .as_array()
            .unwrap()
            .len(),
        1
    );
}

/// PROTOCOL 3.3: a second response to an answered call cannot replace its result.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn late_responses_are_dropped() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(call(&h, "probe.late_answer", json!({})).await, 42);
    assert_eq!(
        call(&h, "probe.echo", json!({"next":true})).await,
        json!({"next":true})
    );
    h.stop().await;
}

/// PROTOCOL 3.4: core.call notifications forget detached children after the caller ends.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn notification_core_call_forgets_detached_child() {
    let w = TestWorld::new();
    let h = w.start().await;
    let parent = call(&h, "probe.forget", json!({})).await;
    let child: Value = tokio::time::timeout(std::time::Duration::from_secs(10), async {
        loop {
            if let Ok(bytes) = std::fs::read(w.0.join("state/forgotten"))
                && let Ok(value) = serde_json::from_slice::<Value>(&bytes)
            {
                break value;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    tokio::time::timeout(std::time::Duration::from_secs(10), async {
        while h.processes().contains_key(parent["pid"].as_str().unwrap()) {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    assert!(h.processes()[child["pid"].as_str().unwrap()].detached);
    assert!(
        h.processes()[child["pid"].as_str().unwrap()]
            .parent
            .is_none()
    );
    call(&h, "core.stop", json!({"pid":child["pid"]})).await;
    assert!(!h.processes().contains_key(child["pid"].as_str().unwrap()));
    h.stop().await;
}

/// Checks the attached-tree half of the shared deadline test.
async fn attached_tree_stops_in_parallel() {
    let w = TestWorld::new();
    let h = w.start().await;
    let root = call(&h, "probe.tree", json!({"depth":2})).await;
    assert_eq!(
        h.processes()
            .values()
            .filter(|p| p.method == "tree")
            .count(),
        3
    );
    let before = std::time::Instant::now();
    call(&h, "core.stop", json!({"pid":root})).await;
    tokio::time::timeout(std::time::Duration::from_secs(2), async {
        while h.processes().values().any(|p| p.method == "tree") {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    assert!(before.elapsed() >= std::time::Duration::from_millis(4900));
    assert!(before.elapsed() < std::time::Duration::from_secs(7));
    h.stop().await;
}

/// PROTOCOL 3.8: a nested call receives its caller's process ID as SAUSAGE_PARENT.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn nested_parent_environment() {
    let w = TestWorld::new();
    let h = w.start().await;
    let result = call(&h, "probe.nested_environment", json!({})).await;
    assert_eq!(result["child"]["parent"], result["caller"]);
    assert_ne!(result["child"]["pid"], result["caller"]);
    h.stop().await;
}

/// PROTOCOL 8 and 3.8: boot services receive core as SAUSAGE_PARENT.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn service_parent_environment() {
    let w = TestWorld::new();
    std::fs::write(
        w.0.join("config/boot.yaml"),
        br#"[{"method":"probe.forgotten","params":{}}]"#,
    )
    .unwrap();
    let h = w.start().await;
    let env: Value =
        serde_json::from_slice(&std::fs::read(w.0.join("state/forgotten")).unwrap()).unwrap();
    assert_eq!(env["parent"], "core");
    assert_eq!(env["plugin"], "probe");
    assert!(h.processes()[env["pid"].as_str().unwrap()].detached);
    h.stop().await;
}

/// PROTOCOL 4: info.json exposes every field of the live process snapshot.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn info_json_has_all_process_fields() {
    let w = TestWorld::new();
    let h = w.start().await;
    let params = json!({"marker":"complete"});
    let pid = call(&h, "probe.serve", params.clone()).await;
    let text = call(
        &h,
        "probe.read",
        json!({"path":format!("/proc/{}/info.json", pid.as_str().unwrap())}),
    )
    .await;
    let info: Value = serde_json::from_str(text.as_str().unwrap()).unwrap();
    let expected = serde_json::to_value(&h.processes()[pid.as_str().unwrap()]).unwrap();
    assert_eq!(info, expected);
    for field in [
        "plugin",
        "version",
        "method",
        "params",
        "parent",
        "detached",
        "start_time",
        "state",
    ] {
        assert!(info.get(field).is_some(), "{field}");
    }
    assert_eq!(info["params"], params);
    assert_eq!(info["parent"], "console");
    assert_eq!(info["state"], "running");
    h.stop().await;
}

/// PROTOCOL 3.6: detached processes persist until stop, kill, or host shutdown explicitly ends them.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn detached_processes_need_explicit_termination() {
    let w = TestWorld::new();
    let h = w.start().await;
    let mut pids = Vec::new();
    for _ in 0..3 {
        let pid = call(
            &h,
            "core.call",
            json!({"method":"probe.serve","params":{},"detach":true}),
        )
        .await;
        assert!(h.processes()[pid.as_str().unwrap()].parent.is_none());
        pids.push(pid);
    }
    call(&h, "probe.echo", json!({})).await;
    for pid in &pids {
        assert!(h.processes().contains_key(pid.as_str().unwrap()));
    }
    call(&h, "core.stop", json!({"pid":pids[0]})).await;
    call(&h, "core.kill", json!({"pid":pids[1]})).await;
    assert!(h.processes().contains_key(pids[2].as_str().unwrap()));
    h.stop().await;
    for pid in pids {
        assert!(!h.processes().contains_key(pid.as_str().unwrap()));
    }
}

/// PROTOCOL 9: restart answers the detached caller, then uses normal stop and status 75.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn core_restart_cleanly_stops_detached_caller() {
    tokio::time::timeout(std::time::Duration::from_secs(30), async {
        let w = TestWorld::new();
        std::fs::write(
            w.0.join("config/boot.yaml"),
            br#"[{"method":"probe.restart","params":{}}]"#,
        )
        .unwrap();
        let output = tokio::process::Command::new(env!("CARGO_BIN_EXE_sausage"))
            .env("SAUSAGE_LOOP", "1")
            .arg("--headless")
            .arg("--cache-dir")
            .arg(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache"))
            .arg("run")
            .arg(&w.0)
            .stdin(std::process::Stdio::null())
            .kill_on_drop(true)
            .output()
            .await
            .unwrap();
        assert_eq!(
            output.status.code(),
            Some(75),
            "{}",
            String::from_utf8_lossy(&output.stderr)
        );
        assert_eq!(
            std::fs::read_to_string(w.0.join("state/restart-answer")).unwrap(),
            "received"
        );
        assert_eq!(
            std::fs::read_to_string(w.0.join("state/restart-stop")).unwrap(),
            "persisted after EOF"
        );
        let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
        let count: i64 = db
            .query_row(
                "SELECT count(*) FROM journal WHERE line = 'restart stop hook finished'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1);
        let killed: i64 = db
            .query_row(
                "SELECT count(*) FROM journal WHERE line = 'kill'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(killed, 0);
    })
    .await
    .unwrap();
}

/// PROTOCOL 9: supervised core.exit from boot uses the restart shutdown path and code 75.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn core_exit_alias_sets_executable_status_75() {
    let w = TestWorld::new();
    std::fs::write(
        w.0.join("config/boot.yaml"),
        br#"[{"method":"core.exit","params":{}}]"#,
    )
    .unwrap();
    let output = tokio::time::timeout(
        std::time::Duration::from_secs(30),
        tokio::process::Command::new(env!("CARGO_BIN_EXE_sausage"))
            .arg("--headless")
            .arg("--cache-dir")
            .arg(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache"))
            .arg("run")
            .arg(&w.0)
            .env("SAUSAGE_LOOP", "1")
            .stdin(std::process::Stdio::null())
            .kill_on_drop(true)
            .output(),
    )
    .await
    .unwrap()
    .unwrap();
    assert_eq!(
        output.status.code(),
        Some(75),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
}