car-server-core 0.50.0

//! car#661: the `auth.*` Parslee-login surface is host-gated, over a real
//! `run_dispatch` WebSocket session.
//!
//! These methods own the daemon's Parslee identity — `logout` clears the active
//! login's tokens, `switch_org` / `switch_account` repoint which identity
//! subsequent inference runs and bills against, `remove_account` drops a stored
//! login. They previously took only the request params, so any authenticated
//! local connection (a registered supervised agent, or any process that read the
//! auth token from `GET /auth-token`) could sign the user out of every
//! Parslee-routed model with one call.
//!
//! Same trust root as `openrouter.*` (#650) and `messaging.*`. This proves a
//! non-host connection is rejected on every one of them, including the reads,
//! and that the host role lifts the gate.

use car_memgine::MemgineEngine;
use car_server_core::{run_dispatch, ServerState, ServerStateConfig};
use futures::{SinkExt, StreamExt};
use std::io::{Read, Write};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{mpsc, Arc, OnceLock};
use std::thread;
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::sync::Mutex;
use tokio_tungstenite::{accept_async, connect_async, tungstenite::Message};

type Ws =
    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;

const HOST_TOKEN: &str = "test-host-token-ccccccccccccccccccccccccccc";

/// Ceiling for every "this must eventually happen" wait in this file.
///
/// Deliberately far larger than the work it covers. A healthy run returns the
/// moment the event lands, so a bigger number costs a passing run nothing — the
/// only thing it buys is not mistaking a loaded machine for a broken daemon.
/// Each of these waits used to carry its own one- or two-second budget, and on a
/// busy operator Mac the budget, not the daemon, is what expired (car#850).
const EVENT_WAIT: std::time::Duration = std::time::Duration::from_secs(30);

fn gated_state(journal_dir: std::path::PathBuf) -> Arc<ServerState> {
    let engine = Arc::new(Mutex::new(MemgineEngine::new(None)));
    let cfg = ServerStateConfig::new(journal_dir).with_shared_memgine(engine);
    let state = Arc::new(ServerState::with_config(cfg));
    // A configured host token is what arms `require_approval_authority`.
    // Without one the daemon is in dev/embedder mode and the gate is a no-op
    // by design (same as permission.* / messaging.* / openrouter.*).
    state
        .install_host_token(HOST_TOKEN.to_string())
        .expect("install host token");
    state
}

async fn spawn_dispatcher(state: Arc<ServerState>, connections: usize) -> SocketAddr {
    let listener = TcpListener::bind(SocketAddr::V4(SocketAddrV4::new(
        Ipv4Addr::new(127, 0, 0, 1),
        0,
    )))
    .await
    .expect("bind loopback");
    let addr = listener.local_addr().expect("local_addr");
    tokio::spawn(async move {
        for _ in 0..connections {
            let (stream, peer) = listener.accept().await.expect("accept");
            let ws = accept_async(stream).await.expect("ws handshake");
            let (write, read) = ws.split();
            let state = state.clone();
            tokio::spawn(async move {
                let _ = run_dispatch(read, Box::pin(write), peer.to_string(), state).await;
            });
        }
    });
    addr
}

async fn call(ws: &mut Ws, id: &str, method: &str, params: serde_json::Value) -> serde_json::Value {
    ws.send(Message::Text(
        serde_json::json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params })
            .to_string()
            .into(),
    ))
    .await
    .expect("send");
    loop {
        let text = ws
            .next()
            .await
            .expect("frame")
            .expect("frame ok")
            .into_text()
            .expect("text")
            .to_string();
        let value: serde_json::Value = serde_json::from_str(&text).expect("parse");
        if value.get("id").and_then(serde_json::Value::as_str) == Some(id) {
            return value;
        }
    }
}

fn run_isolated_contract(test_name: &str, sentinel: &str) -> bool {
    if std::env::var_os(sentinel).is_some() {
        return false;
    }
    let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
        .arg("--exact")
        .arg(test_name)
        .arg("--nocapture")
        .env(sentinel, "1")
        .env_remove("CAR_SECRETS_FILE_DIR")
        .env_remove(car_home::ENV_VAR)
        .env_remove(car_auth::PARSLEE_ACCESS_TOKEN_KEY)
        .env_remove(car_auth::PARSLEE_REFRESH_TOKEN_KEY)
        .env_remove(car_auth::PARSLEE_EXPIRES_AT_KEY)
        .env_remove(car_auth::PARSLEE_API_BASE_KEY)
        .env("CAR_NO_INFERENCE_WORKER", "1")
        .output()
        .expect("spawn isolated contract test");
    assert!(
        output.status.success(),
        "isolated {test_name} failed\nstdout:\n{}\nstderr:\n{}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr),
    );
    true
}

#[test]
fn isolated_auth_surface_child_sanitizes_parslee_process_overrides() {
    const LAUNCHER: &str = "CAR_TASK3_AUTH_ENV_LAUNCHER";
    const INNER: &str = "CAR_TASK3_AUTH_ENV_INNER";
    const TEST_NAME: &str = "isolated_auth_surface_child_sanitizes_parslee_process_overrides";

    if std::env::var_os(INNER).is_some() {
        assert!(
            std::env::var_os(car_auth::PARSLEE_ACCESS_TOKEN_KEY).is_none(),
            "the isolated auth child inherited PARSLEE_ACCESS_TOKEN"
        );
        assert!(
            std::env::var_os(car_auth::PARSLEE_API_BASE_KEY).is_none(),
            "the isolated auth child inherited PARSLEE_API_BASE"
        );
        return;
    }

    if std::env::var_os(LAUNCHER).is_some() {
        assert!(run_isolated_contract(TEST_NAME, INNER));
        return;
    }

    let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
        .arg("--exact")
        .arg(TEST_NAME)
        .arg("--nocapture")
        .env(LAUNCHER, "1")
        .env(car_auth::PARSLEE_ACCESS_TOKEN_KEY, "must-not-reach-child")
        .env(car_auth::PARSLEE_API_BASE_KEY, "http://127.0.0.1:9")
        .output()
        .expect("spawn hostile auth-env launcher");
    assert!(
        output.status.success(),
        "hostile auth-env isolation failed\nstdout:\n{}\nstderr:\n{}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr),
    );
}

fn spawn_session_mock() -> String {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind session mock");
    let address = listener.local_addr().expect("session mock address");
    thread::spawn(move || {
        let (mut stream, _) = listener.accept().expect("accept session request");
        read_http_request(&mut stream);
        let body =
            r#"{"Authenticated":true,"Account":{"Id":"account-1","Email":"person@example.test"}}"#;
        let response = format!(
            "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
            body.len(), body
        );
        stream
            .write_all(response.as_bytes())
            .expect("write session response");
    });
    format!("http://{address}")
}

fn spawn_session_and_mobile_mock() -> (String, std::sync::mpsc::Receiver<()>) {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind activation mock");
    let address = listener.local_addr().expect("activation mock address");
    let (registered_tx, registered_rx) = std::sync::mpsc::channel();
    thread::spawn(move || {
        let (mut status_stream, _) = listener.accept().expect("accept session request");
        read_http_request(&mut status_stream);
        let body = r#"{"Authenticated":true,"Account":{"Id":"account-1","Email":"person@example.test"},"ActiveOrganization":"org-1"}"#;
        let response = format!(
            "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
            body.len(), body
        );
        status_stream
            .write_all(response.as_bytes())
            .expect("write session response");

        let (mut registration_stream, _) = listener.accept().expect("accept mobile registration");
        read_http_request(&mut registration_stream);
        registration_stream
            .write_all(b"HTTP/1.1 204 No Content\r\ncontent-length: 0\r\nconnection: close\r\n\r\n")
            .expect("write registration response");
        registered_tx.send(()).expect("signal mobile registration");
    });
    (format!("http://{address}"), registered_rx)
}

fn spawn_repeated_session_and_mobile_mock() -> (
    String,
    std::sync::mpsc::Receiver<()>,
    std::sync::mpsc::Receiver<Vec<String>>,
) {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind parslee.auth mock");
    let address = listener.local_addr().expect("parslee.auth mock address");
    let (registered_tx, registered_rx) = std::sync::mpsc::channel();
    let (requests_tx, requests_rx) = std::sync::mpsc::channel();
    thread::spawn(move || {
        let mut requests = Vec::new();
        for _ in 0..3 {
            let (mut stream, _) = listener.accept().expect("accept parslee.auth request");
            let request = read_http_request(&mut stream);
            let request_line = request.lines().next().unwrap_or_default().to_string();
            requests.push(request_line.clone());
            if request_line.starts_with("GET /connect/session ") {
                let body = r#"{"Authenticated":true,"Account":{"Id":"account-1","Email":"person@example.test"},"ActiveOrganization":"org-1"}"#;
                let response = format!(
                    "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
                    body.len(), body
                );
                stream
                    .write_all(response.as_bytes())
                    .expect("write parslee.auth session response");
            } else if request_line.starts_with("POST /mobile/runtimes ") {
                stream
                    .write_all(
                        b"HTTP/1.1 204 No Content\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
                    )
                    .expect("write parslee.auth registration response");
                registered_tx
                    .send(())
                    .expect("signal parslee.auth registration");
            } else {
                panic!("unexpected request to loopback auth mock: {request_line}");
            }
        }

        listener
            .set_nonblocking(true)
            .expect("make parslee.auth mock nonblocking");
        let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500);
        while std::time::Instant::now() < deadline {
            match listener.accept() {
                Ok((mut stream, _)) => {
                    let request = read_http_request(&mut stream);
                    requests.push(request.lines().next().unwrap_or_default().to_string());
                }
                Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                    thread::sleep(std::time::Duration::from_millis(5));
                }
                Err(error) => panic!("accept duplicate registration probe: {error}"),
            }
        }
        requests_tx
            .send(requests)
            .expect("publish observed auth requests");
    });
    (format!("http://{address}"), registered_rx, requests_rx)
}

async fn wait_for_terminal_proof(ws: &mut Ws, attempt_id: &str) -> serde_json::Value {
    let deadline = std::time::Instant::now() + EVENT_WAIT;
    let mut poll = 0_usize;
    while std::time::Instant::now() < deadline {
        let response = call(
            ws,
            &format!("proof-{poll}"),
            "auth.completion_status",
            serde_json::json!({ "attempt_id": attempt_id }),
        )
        .await;
        if response["result"]["state"] != "pending" {
            return response["result"].clone();
        }
        poll += 1;
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    panic!("attempt `{attempt_id}` did not publish a terminal proof");
}

fn auth_env_lock() -> &'static tokio::sync::Mutex<()> {
    static LOCK: OnceLock<tokio::sync::Mutex<()>> = OnceLock::new();
    LOCK.get_or_init(|| tokio::sync::Mutex::new(()))
}

async fn authenticate_host(ws: &mut Ws) {
    let auth = call(
        ws,
        "host-auth",
        "session.auth",
        serde_json::json!({ "host_token": HOST_TOKEN }),
    )
    .await;
    assert_eq!(
        auth["result"]["role"], "host",
        "host auth should succeed: {auth}"
    );
    negotiate(ws).await;
}

async fn negotiate(ws: &mut Ws) {
    let handshake = call(
        ws,
        "protocol-v2",
        "server.handshake",
        serde_json::json!({ "protocol_version": car_proto::PROTOCOL_VERSION }),
    )
    .await;
    assert_eq!(
        handshake["result"]["protocol_version"],
        car_proto::PROTOCOL_VERSION,
        "protocol handshake should succeed: {handshake}"
    );
}

fn read_http_request(stream: &mut std::net::TcpStream) -> String {
    let mut bytes = Vec::new();
    let mut chunk = [0_u8; 1024];
    loop {
        let read = stream.read(&mut chunk).expect("read HTTP request");
        assert!(read > 0, "HTTP client closed before request headers");
        bytes.extend_from_slice(&chunk[..read]);
        let Some(headers_end) = bytes.windows(4).position(|window| window == b"\r\n\r\n") else {
            continue;
        };
        let headers = String::from_utf8_lossy(&bytes[..headers_end]);
        let content_length = headers
            .lines()
            .find_map(|line| {
                let (name, value) = line.split_once(':')?;
                name.eq_ignore_ascii_case("content-length")
                    .then(|| value.trim().parse::<usize>().ok())
                    .flatten()
            })
            .unwrap_or_default();
        if bytes.len() >= headers_end + 4 + content_length {
            return String::from_utf8_lossy(&bytes).into_owned();
        }
    }
}

struct OAuthMock {
    base: String,
    token_seen: mpsc::Receiver<()>,
    release_session: mpsc::Sender<()>,
    token_requests: Arc<AtomicUsize>,
}

fn spawn_oauth_mock() -> OAuthMock {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind OAuth mock");
    let address = listener.local_addr().expect("mock address");
    let (token_seen_tx, token_seen) = mpsc::channel();
    let (release_session, release_session_rx) = mpsc::channel();
    let token_requests = Arc::new(AtomicUsize::new(0));
    let token_requests_thread = token_requests.clone();
    thread::spawn(move || {
        let (mut token_stream, _) = listener.accept().expect("accept token request");
        read_http_request(&mut token_stream);
        token_requests_thread.fetch_add(1, Ordering::SeqCst);
        let body = r#"{"access_token":"access-1","refresh_token":"refresh-1","expires_in":3600,"token_type":"Bearer"}"#;
        let response = format!(
            "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
            body.len(), body
        );
        token_stream
            .write_all(response.as_bytes())
            .expect("write token response");
        token_seen_tx.send(()).expect("notify token request");

        let (mut failed_session_stream, _) =
            listener.accept().expect("accept first session request");
        read_http_request(&mut failed_session_stream);
        let failure = "temporary session failure";
        let failed_response = format!(
            "HTTP/1.1 503 Service Unavailable\r\ncontent-type: text/plain\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
            failure.len(), failure
        );
        failed_session_stream
            .write_all(failed_response.as_bytes())
            .expect("write transient session failure");

        let (mut session_stream, _) = listener.accept().expect("accept retry session request");
        read_http_request(&mut session_stream);
        release_session_rx.recv().expect("release session response");
        let body = r#"{"Account":{"Id":"account-1","Email":"person@example.test"}}"#;
        let response = format!(
            "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
            body.len(), body
        );
        session_stream
            .write_all(response.as_bytes())
            .expect("write session response");
    });
    OAuthMock {
        base: format!("http://{address}"),
        token_seen,
        release_session,
        token_requests,
    }
}

struct RejectingOAuthMock {
    base: String,
    /// Every request the daemon actually sent, so the count is the assertion
    /// and not a by-product of how long the mock happened to stay up.
    token_requests: Arc<AtomicUsize>,
    /// One message per served request: lets the test wait for the exchange to
    /// happen instead of budgeting wall-clock time for it.
    token_seen: mpsc::Receiver<()>,
    stop: Arc<AtomicBool>,
    handle: thread::JoinHandle<()>,
}

fn spawn_rejecting_oauth_mock() -> RejectingOAuthMock {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind OAuth mock");
    listener.set_nonblocking(true).unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    let token_requests = Arc::new(AtomicUsize::new(0));
    let requests = token_requests.clone();
    let stop = Arc::new(AtomicBool::new(false));
    let stop_serving = stop.clone();
    let (token_seen_tx, token_seen) = mpsc::channel();
    // Serve until the test stops us. This loop used to run against a one-second
    // wall clock started here — before the dispatcher had even booted — so on a
    // loaded machine the mock stopped listening before the single legitimate
    // token exchange arrived and the count read 0 rather than 1 (car#850).
    let handle = thread::spawn(move || {
        while !stop_serving.load(Ordering::SeqCst) {
            match listener.accept() {
                Ok((mut stream, _)) => {
                    stream
                        .set_nonblocking(false)
                        .expect("make accepted OAuth stream blocking");
                    read_http_request(&mut stream);
                    requests.fetch_add(1, Ordering::SeqCst);
                    let body = "authorization code rejected";
                    let response = format!(
                        "HTTP/1.1 400 Bad Request\r\ncontent-type: text/plain\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
                        body.len(),
                        body
                    );
                    stream.write_all(response.as_bytes()).unwrap();
                    // A dropped receiver (the test already failed elsewhere)
                    // must not turn into a second panic from this thread.
                    let _ = token_seen_tx.send(());
                }
                Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                    thread::sleep(std::time::Duration::from_millis(5));
                }
                Err(error) => panic!("accept OAuth request: {error}"),
            }
        }
    });
    RejectingOAuthMock {
        base,
        token_requests,
        token_seen,
        stop,
        handle,
    }
}

#[tokio::test]
async fn parslee_login_management_requires_host_role() {
    let _env_guard = auth_env_lock().lock().await;
    let journal = TempDir::new().unwrap();
    // Redirect `~/` AND the secret store so nothing here can touch a developer's
    // real token store. `HOME` alone is insufficient: on Windows home resolution
    // uses `USERPROFILE`, and the Parslee login token lives in the OS credential
    // store (Windows Credential Manager / macOS keychain / libsecret), which no
    // home redirect isolates — so on any machine with a real `car auth login`
    // the `authenticated: false` assertion below would spuriously fail (it does
    // on a logged-in Windows dev box; clean CI has no login so it passed there).
    // `CAR_SECRETS_FILE_DIR` routes `SecretStore` to an isolated (empty) dir on
    // every platform — the crate's documented test seam, process-global and thus
    // safe under nextest's process-per-test model. NB: the seam is honored only
    // under `cfg!(debug_assertions)` (car-secrets `file_backend_dir`), so this
    // isolation applies to `cargo test`/nextest (debug) but NOT `--release`; a
    // release test run on a logged-in box would read the real keychain again.
    // SAFETY: this test binary is the only thing in this process.
    let fake_home = TempDir::new().unwrap();
    let fake_secrets = TempDir::new().unwrap();
    std::env::set_var("HOME", fake_home.path());
    std::env::set_var("USERPROFILE", fake_home.path());
    // `CAR_HOME` outranks both `HOME` and `USERPROFILE` in the state root, so
    // pinning them alone no longer isolates — a developer who exports it would
    // send the token lookup back at their real state root.
    std::env::remove_var(car_home::ENV_VAR);
    std::env::set_var("CAR_SECRETS_FILE_DIR", fake_secrets.path());

    let state = gated_state(journal.path().to_path_buf());
    let addr = spawn_dispatcher(state, 1).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}"))
        .await
        .expect("connect");
    negotiate(&mut ws).await;

    // 1. A non-host connection is rejected on every auth.* method — the
    //    identity mutators AND the reads that expose stored-login/attempt state.
    for (i, (method, params)) in [
        ("auth.authority_hint", serde_json::json!({})),
        ("auth.snapshot", serde_json::json!({})),
        (
            "auth.completion_status",
            serde_json::json!({ "attempt_id": "attempt_x" }),
        ),
        ("auth.status", serde_json::json!({})),
        ("auth.accounts", serde_json::json!({})),
        ("auth.logout", serde_json::json!({})),
        ("auth.switch_org", serde_json::json!({ "org_id": "org_x" })),
        (
            "auth.switch_account",
            serde_json::json!({ "account_id": "acct_x" }),
        ),
        (
            "auth.remove_account",
            serde_json::json!({ "account_id": "acct_x" }),
        ),
        ("auth.start", serde_json::json!({})),
        ("auth.complete", serde_json::json!({ "code": "x" })),
    ]
    .into_iter()
    .enumerate()
    {
        let resp = call(&mut ws, &format!("n{i}"), method, params).await;
        assert!(
            resp.get("error").is_some(),
            "{method} must REJECT a non-host caller, got: {resp}"
        );
        assert!(
            resp.get("result").is_none(),
            "{method} must not return a result to a non-host caller"
        );
        // The rejection must be the authority gate, not an incidental
        // parameter/validation error that would vanish with valid params.
        let message = resp["error"]["message"].as_str().unwrap_or_default();
        assert!(
            message.contains("host-management role"),
            "{method} must be refused BY THE GATE, got: {message}"
        );
    }

    // 2. Take the host-management role with the per-launch host token.
    authenticate_host(&mut ws).await;

    // 3. As host the gate is lifted. `auth.status` is the safe one to assert
    //    on: it reads local token presence, starts no OAuth flow, and with no
    //    stored login simply reports `authenticated: false`.
    let status = call(&mut ws, "h1", "auth.status", serde_json::json!({})).await;
    assert!(
        status.get("error").is_none(),
        "host auth.status should pass the gate: {status}"
    );
    assert_eq!(
        status["result"]["authenticated"], false,
        "fixture HOME has no stored login: {status}"
    );
}

#[tokio::test]
async fn dropped_auth_complete_reply_persists_attempt_proof_without_replaying_code() {
    let _env_guard = auth_env_lock().lock().await;
    let journal = TempDir::new().unwrap();
    let secrets = TempDir::new().unwrap();
    std::env::set_var("CAR_SECRETS_FILE_DIR", secrets.path());
    let OAuthMock {
        base,
        token_seen,
        release_session,
        token_requests,
    } = spawn_oauth_mock();
    let state = gated_state(journal.path().to_path_buf());
    let addr = spawn_dispatcher(state, 2).await;

    let (mut first, _) = connect_async(format!("ws://{addr}"))
        .await
        .expect("connect first host");
    authenticate_host(&mut first).await;
    let started = call(
        &mut first,
        "start",
        "auth.start",
        serde_json::json!({
            "api_base": base.clone(),
            "redirect_uri": "http://127.0.0.1/callback"
        }),
    )
    .await;
    let attempt_id = started["result"]["attempt_id"]
        .as_str()
        .expect("attempt id")
        .to_string();
    let verifier = started["result"]["verifier"]
        .as_str()
        .expect("verifier")
        .to_string();

    first
        .send(Message::Text(
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": "complete",
                "method": "auth.complete",
                "params": {
                    "api_base": base.clone(),
                    "redirect_uri": "http://127.0.0.1/callback",
                    "code": "one-time-code",
                    "verifier": verifier,
                    "attempt_id": attempt_id.clone(),
                }
            })
            .to_string()
            .into(),
        ))
        .await
        .expect("send complete");
    tokio::task::spawn_blocking(move || {
        token_seen
            .recv_timeout(EVENT_WAIT)
            .expect("completion must exchange the code before disconnect");
    })
    .await
    .expect("token wait task");
    drop(first);

    let (mut second, _) = connect_async(format!("ws://{addr}"))
        .await
        .expect("connect reconciliation host");
    authenticate_host(&mut second).await;
    let release = tokio::spawn(async move {
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        release_session
            .send(())
            .expect("release completion session response");
    });
    let proof = wait_for_terminal_proof(&mut second, &attempt_id).await;
    release.await.unwrap();

    assert_eq!(proof["attempt_id"], attempt_id);
    assert_eq!(
        proof["state"], "complete",
        "a proof request during redemption must wait for terminal publication"
    );
    assert_eq!(proof["account_id"], "account-1");
    assert_eq!(
        token_requests.load(Ordering::SeqCst),
        1,
        "the one-time code must not be replayed"
    );
    std::env::remove_var("CAR_SECRETS_FILE_DIR");
}

#[tokio::test]
async fn auth_complete_accepts_before_session_validation_and_publishes_through_proof() {
    let _env_guard = auth_env_lock().lock().await;
    let journal = TempDir::new().unwrap();
    let secrets = TempDir::new().unwrap();
    std::env::set_var("CAR_SECRETS_FILE_DIR", secrets.path());
    let OAuthMock {
        base,
        token_seen,
        release_session,
        token_requests,
    } = spawn_oauth_mock();
    let state = gated_state(journal.path().to_path_buf());
    let addr = spawn_dispatcher(state, 1).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    authenticate_host(&mut ws).await;

    let started = call(
        &mut ws,
        "start",
        "auth.start",
        serde_json::json!({
            "api_base": base.clone(),
            "redirect_uri": "http://127.0.0.1/callback"
        }),
    )
    .await;
    let attempt_id = started["result"]["attempt_id"]
        .as_str()
        .expect("attempt id")
        .to_string();

    ws.send(Message::Text(
        serde_json::json!({
            "jsonrpc": "2.0",
            "id": "complete",
            "method": "auth.complete",
            "params": {
                "api_base": base,
                "redirect_uri": "http://127.0.0.1/callback",
                "code": "one-time-code",
                "verifier": started["result"]["verifier"],
                "attempt_id": attempt_id,
            }
        })
        .to_string()
        .into(),
    ))
    .await
    .unwrap();
    tokio::task::spawn_blocking(move || {
        token_seen
            .recv_timeout(EVENT_WAIT)
            .expect("worker should exchange the code");
    })
    .await
    .unwrap();

    let accepted = tokio::time::timeout(EVENT_WAIT, ws.next())
        .await
        .expect("auth.complete must acknowledge before session validation")
        .expect("accepted frame")
        .expect("accepted frame ok");
    let accepted: serde_json::Value = serde_json::from_str(&accepted.into_text().unwrap()).unwrap();
    assert_eq!(
        accepted["result"],
        serde_json::json!({
            "state": "accepted",
            "attempt_id": started["result"]["attempt_id"],
        }),
        "the direct reply is only an acceptance receipt; proof remains authoritative"
    );
    let claimed = call(
        &mut ws,
        "claimed-proof",
        "auth.completion_status",
        serde_json::json!({ "attempt_id": started["result"]["attempt_id"] }),
    )
    .await;
    assert_eq!(
        claimed["result"]["phase"], "redeeming",
        "accepted must mean the exact attempt fence is already durably claimed: {claimed}"
    );

    release_session.send(()).unwrap();
    let proof =
        wait_for_terminal_proof(&mut ws, started["result"]["attempt_id"].as_str().unwrap()).await;
    assert_eq!(proof["state"], "complete", "{proof}");
    assert_eq!(proof["account_id"], "account-1", "{proof}");
    assert_eq!(
        token_requests.load(Ordering::SeqCst),
        1,
        "proof reconciliation must not replay the one-time code"
    );
    std::env::remove_var("CAR_SECRETS_FILE_DIR");
}

#[tokio::test]
async fn stale_or_missing_attempt_is_rejected_before_token_exchange() {
    let _env_guard = auth_env_lock().lock().await;
    let journal = TempDir::new().unwrap();
    let secrets = TempDir::new().unwrap();
    std::env::set_var("CAR_SECRETS_FILE_DIR", secrets.path());
    let token_listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    token_listener.set_nonblocking(true).unwrap();
    let api_base = format!("http://{}", token_listener.local_addr().unwrap());
    let state = gated_state(journal.path().to_path_buf());
    let addr = spawn_dispatcher(state, 1).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    authenticate_host(&mut ws).await;

    let start_a = call(
        &mut ws,
        "start-a",
        "auth.start",
        serde_json::json!({
            "api_base": api_base.clone(),
            "redirect_uri": "http://127.0.0.1/callback-a"
        }),
    )
    .await;
    let start_b = call(
        &mut ws,
        "start-b",
        "auth.start",
        serde_json::json!({
            "api_base": api_base.clone(),
            "redirect_uri": "http://127.0.0.1/callback-b"
        }),
    )
    .await;
    let attempt_a = start_a["result"]["attempt_id"].as_str().unwrap();
    let attempt_b = start_b["result"]["attempt_id"].as_str().unwrap();

    let stale = call(
        &mut ws,
        "complete-a",
        "auth.complete",
        serde_json::json!({
            "api_base": api_base.clone(),
            "redirect_uri": "http://127.0.0.1/callback-a",
            "code": "code-a",
            "verifier": start_a["result"]["verifier"],
            "attempt_id": attempt_a,
        }),
    )
    .await;
    assert_eq!(stale["error"]["code"], -32603, "{stale}");
    assert!(
        stale["error"]["message"]
            .as_str()
            .is_some_and(|message| message.contains("stale")),
        "a failed preclaim must never produce accepted or begin redemption: {stale}"
    );

    let missing = call(
        &mut ws,
        "complete-missing",
        "auth.complete",
        serde_json::json!({
            "api_base": api_base,
            "redirect_uri": "http://127.0.0.1/callback-a",
            "code": "code-a",
            "verifier": start_a["result"]["verifier"],
        }),
    )
    .await;
    assert!(
        missing["error"]["message"]
            .as_str()
            .is_some_and(|message| message.contains("attempt_id")),
        "legacy completion without an attempt must fail closed: {missing}"
    );

    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    assert!(
        matches!(
            token_listener.accept(),
            Err(error) if error.kind() == std::io::ErrorKind::WouldBlock
        ),
        "neither rejected request may reach /connect/token"
    );

    let stale_status = call(
        &mut ws,
        "status-a",
        "auth.completion_status",
        serde_json::json!({ "attempt_id": attempt_a }),
    )
    .await;
    assert_eq!(stale_status["result"]["state"], "stale");
    let pending_status = call(
        &mut ws,
        "status-b",
        "auth.completion_status",
        serde_json::json!({ "attempt_id": attempt_b }),
    )
    .await;
    assert_eq!(pending_status["result"]["state"], "pending");
    assert_eq!(pending_status["result"]["phase"], "awaiting_callback");
    assert!(pending_status["result"]["expires_at_unix_ms"].is_number());
    std::env::remove_var("CAR_SECRETS_FILE_DIR");
}

#[tokio::test]
async fn duplicate_complete_claims_redeem_one_code_once() {
    let _env_guard = auth_env_lock().lock().await;
    let journal = TempDir::new().unwrap();
    let secrets = TempDir::new().unwrap();
    std::env::set_var("CAR_SECRETS_FILE_DIR", secrets.path());
    let RejectingOAuthMock {
        base: api_base,
        token_requests,
        token_seen,
        stop: stop_mock,
        handle: mock_thread,
    } = spawn_rejecting_oauth_mock();
    let state = gated_state(journal.path().to_path_buf());
    let addr = spawn_dispatcher(state, 1).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    authenticate_host(&mut ws).await;
    let started = call(
        &mut ws,
        "start",
        "auth.start",
        serde_json::json!({
            "api_base": api_base.clone(),
            "redirect_uri": "http://127.0.0.1/callback"
        }),
    )
    .await;
    let params = serde_json::json!({
        "api_base": api_base,
        "redirect_uri": "http://127.0.0.1/callback",
        "code": "one-time-code",
        "verifier": started["result"]["verifier"],
        "attempt_id": started["result"]["attempt_id"],
    });

    for id in ["complete-1", "complete-2"] {
        ws.send(Message::Text(
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": id,
                "method": "auth.complete",
                "params": params,
            })
            .to_string()
            .into(),
        ))
        .await
        .unwrap();
    }
    let first: serde_json::Value =
        serde_json::from_str(&ws.next().await.unwrap().unwrap().into_text().unwrap()).unwrap();
    let second: serde_json::Value =
        serde_json::from_str(&ws.next().await.unwrap().unwrap().into_text().unwrap()).unwrap();
    let responses = [&first, &second];
    assert_eq!(
        responses
            .iter()
            .filter(|response| response["result"]["state"] == "accepted")
            .count(),
        1,
        "exactly one request may durably claim the attempt: {responses:?}"
    );
    assert_eq!(
        responses
            .iter()
            .filter(|response| {
                response["error"]["code"] == -32603
                    && response["error"]["message"]
                        .as_str()
                        .is_some_and(|message| message.contains("already being redeemed"))
            })
            .count(),
        1,
        "the duplicate must fail before a second redemption begins: {responses:?}"
    );

    // Wait for the winner's exchange to actually reach the mock. The generous
    // bound is a ceiling on a machine that is merely slow, not a budget the
    // happy path has to beat: a healthy run returns as soon as the request
    // lands.
    let token_seen = tokio::task::spawn_blocking(move || token_seen.recv_timeout(EVENT_WAIT))
        .await
        .unwrap();
    token_seen.expect("the accepted claim must exchange its code exactly once");

    // The mock rejects that code, so the claiming worker fails terminally.
    // Waiting for the published terminal proof is what closes the window on a
    // second redemption — after it, no further exchange can legitimately be in
    // flight — instead of sleeping and hoping.
    let proof =
        wait_for_terminal_proof(&mut ws, started["result"]["attempt_id"].as_str().unwrap()).await;
    assert_eq!(
        proof["state"], "failed",
        "a rejected code must fail the attempt rather than sign anyone in: {proof}"
    );

    stop_mock.store(true, Ordering::SeqCst);
    tokio::task::spawn_blocking(move || mock_thread.join().unwrap())
        .await
        .unwrap();
    assert_eq!(
        token_requests.load(Ordering::SeqCst),
        1,
        "the durable worker claim must allow exactly one token exchange"
    );
    std::env::remove_var("CAR_SECRETS_FILE_DIR");
}

#[tokio::test]
async fn awaiting_callback_can_complete_on_a_new_server_instance() {
    let _env_guard = auth_env_lock().lock().await;
    let journal_a = TempDir::new().unwrap();
    let journal_b = TempDir::new().unwrap();
    let secrets = TempDir::new().unwrap();
    std::env::set_var("CAR_SECRETS_FILE_DIR", secrets.path());
    let OAuthMock {
        base,
        token_seen,
        release_session,
        token_requests,
    } = spawn_oauth_mock();
    let addr_a = spawn_dispatcher(gated_state(journal_a.path().to_path_buf()), 1).await;
    let addr_b = spawn_dispatcher(gated_state(journal_b.path().to_path_buf()), 1).await;
    let (mut first, _) = connect_async(format!("ws://{addr_a}")).await.unwrap();
    let (mut second, _) = connect_async(format!("ws://{addr_b}")).await.unwrap();
    authenticate_host(&mut first).await;
    authenticate_host(&mut second).await;

    let started = call(
        &mut first,
        "start",
        "auth.start",
        serde_json::json!({
            "api_base": base.clone(),
            "redirect_uri": "http://127.0.0.1/callback"
        }),
    )
    .await;
    second
        .send(Message::Text(
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": "complete",
                "method": "auth.complete",
                "params": {
                    "api_base": base,
                    "redirect_uri": "http://127.0.0.1/callback",
                    "code": "one-time-code",
                    "verifier": started["result"]["verifier"],
                    "attempt_id": started["result"]["attempt_id"],
                }
            })
            .to_string()
            .into(),
        ))
        .await
        .unwrap();
    tokio::task::spawn_blocking(move || token_seen.recv_timeout(EVENT_WAIT).unwrap())
        .await
        .unwrap();
    release_session.send(()).unwrap();
    let accepted: serde_json::Value =
        serde_json::from_str(&second.next().await.unwrap().unwrap().into_text().unwrap()).unwrap();

    assert_eq!(accepted["result"]["state"], "accepted", "{accepted}");
    let proof = wait_for_terminal_proof(
        &mut second,
        started["result"]["attempt_id"].as_str().unwrap(),
    )
    .await;
    assert_eq!(proof["state"], "complete", "{proof}");
    assert_eq!(token_requests.load(Ordering::SeqCst), 1);
    std::env::remove_var("CAR_SECRETS_FILE_DIR");
}

#[cfg(unix)]
#[tokio::test]
async fn disconnected_logout_finishes_without_leaking_its_reply_to_reconnect() {
    use std::fs::OpenOptions;
    use std::os::fd::AsRawFd;

    let _env_guard = auth_env_lock().lock().await;
    let journal = TempDir::new().unwrap();
    let secrets = TempDir::new().unwrap();
    let auth_lock_path = secrets.path().join("auth.lock");
    std::env::set_var("CAR_SECRETS_FILE_DIR", secrets.path());
    std::env::set_var("CAR_AUTH_LOCK_PATH", &auth_lock_path);

    let seed = car_auth::TokenSet {
        access_token: "seed-access".into(),
        refresh_token: "seed-refresh".into(),
        expires_in: 3_600,
        token_type: "Bearer".into(),
    };
    car_auth::commit_login(
        "https://api.example.test",
        &seed,
        r#"{"Account":{"Id":"seed-account","Email":"seed@example.test"}}"#,
        None,
    )
    .await
    .unwrap();
    assert!(car_auth::local_auth_snapshot().await.unwrap().authenticated);

    let lock_file = OpenOptions::new()
        .create(true)
        .truncate(false)
        .read(true)
        .write(true)
        .open(&auth_lock_path)
        .unwrap();
    assert_eq!(
        unsafe { libc::flock(lock_file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) },
        0,
        "test must hold the cross-process auth lock"
    );

    // Occupy the in-process coordinator behind the held file lock. The logout
    // request then waits on the async coordinator guard rather than entering a
    // spawn_blocking call of its own, making disconnect cancellation
    // deterministic under the old connection-owned design.
    let blocker = tokio::spawn(car_auth::local_auth_snapshot());
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    assert!(!blocker.is_finished(), "fixture must hold the coordinator");

    let state = gated_state(journal.path().to_path_buf());
    let addr = spawn_dispatcher(state, 2).await;
    let (mut first, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    authenticate_host(&mut first).await;
    first
        .send(Message::Text(
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": "same-id",
                "method": "auth.logout",
                "params": {},
            })
            .to_string()
            .into(),
        ))
        .await
        .unwrap();
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    first.close(None).await.unwrap();
    drop(first);

    assert_eq!(
        unsafe { libc::flock(lock_file.as_raw_fd(), libc::LOCK_UN) },
        0
    );
    blocker.await.unwrap().unwrap();

    let signed_out = tokio::time::timeout(EVENT_WAIT, async {
        loop {
            let snapshot = car_auth::local_auth_snapshot().await.unwrap();
            if !snapshot.authenticated {
                break snapshot;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
    })
    .await
    .expect("daemon-owned logout must finish after disconnect");
    assert!(!signed_out.authenticated);

    let (mut second, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    authenticate_host(&mut second).await;
    let response = call(
        &mut second,
        "same-id",
        "auth.snapshot",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(
        response,
        serde_json::json!({
            "jsonrpc": "2.0",
            "id": "same-id",
            "result": { "authenticated": false },
        }),
        "the old connection's logout reply must not leak or replay"
    );

    std::env::remove_var("CAR_AUTH_LOCK_PATH");
    std::env::remove_var("CAR_SECRETS_FILE_DIR");
}

#[tokio::test]
async fn retry_flag_is_the_only_way_out_of_keychain_cooldown() {
    const SENTINEL: &str = "CAR_TASK3_RETRY_CONTRACT_CHILD";
    if run_isolated_contract(
        "retry_flag_is_the_only_way_out_of_keychain_cooldown",
        SENTINEL,
    ) {
        return;
    }

    let _env_guard = auth_env_lock().lock().await;
    let journal = TempDir::new().unwrap();
    let fake_home = TempDir::new().unwrap();
    let secret_parent = TempDir::new().unwrap();
    let blocked_secret_dir = secret_parent.path().join("blocked");
    std::fs::write(&blocked_secret_dir, b"not a directory").unwrap();
    std::env::set_var("HOME", fake_home.path());
    std::env::set_var("USERPROFILE", fake_home.path());
    std::env::set_var("CAR_SECRETS_FILE_DIR", &blocked_secret_dir);
    std::env::set_var(
        "CAR_AUTH_LOCK_PATH",
        secret_parent.path().join("auth-coordinator.lock"),
    );

    let state = gated_state(journal.path().to_path_buf());
    let addr = spawn_dispatcher(state, 1).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    authenticate_host(&mut ws).await;

    let before = car_secrets::secret_store_activity();
    let denied = call(&mut ws, "denied", "auth.status", serde_json::json!({})).await;
    assert!(
        denied["error"]["message"]
            .as_str()
            .is_some_and(|message| message.contains("file backend")),
        "the physical read must surface its terminal store failure: {denied}"
    );
    let after_denied = car_secrets::secret_store_activity();
    assert_eq!(after_denied.get_attempts, before.get_attempts + 1);

    let cooldown = call(&mut ws, "cooldown", "auth.status", serde_json::json!({})).await;
    assert!(
        cooldown["error"]["message"]
            .as_str()
            .is_some_and(|message| message.contains("in cooldown")),
        "ordinary status must remain in cooldown: {cooldown}"
    );
    assert_eq!(
        car_secrets::secret_store_activity(),
        after_denied,
        "cooldown must not start another physical read"
    );

    std::fs::remove_file(&blocked_secret_dir).unwrap();
    std::fs::create_dir(&blocked_secret_dir).unwrap();
    let api_base = spawn_session_mock();
    car_auth::commit_login(
        &api_base,
        &car_auth::TokenSet {
            access_token: "access-1".into(),
            refresh_token: "refresh-1".into(),
            expires_in: 3_600,
            token_type: "Bearer".into(),
        },
        r#"{"Account":{"Id":"account-1","Email":"person@example.test"}}"#,
        None,
    )
    .await
    .unwrap();
    let before_retry = car_secrets::secret_store_activity();

    let retry = call(
        &mut ws,
        "retry",
        "auth.status",
        serde_json::json!({ "retry_keychain_access": true }),
    )
    .await;
    assert_eq!(retry["result"]["authenticated"], true, "{retry}");
    assert_eq!(
        car_secrets::secret_store_activity().get_attempts,
        before_retry.get_attempts + 1,
        "the explicit retry must start exactly one new physical read"
    );
}

#[tokio::test]
async fn explicit_status_activates_identity_and_mobile_registration() {
    const SENTINEL: &str = "CAR_TASK3_EXPLICIT_ACTIVATION_CHILD";
    if run_isolated_contract(
        "explicit_status_activates_identity_and_mobile_registration",
        SENTINEL,
    ) {
        return;
    }

    let _env_guard = auth_env_lock().lock().await;
    let journal = TempDir::new().unwrap();
    let secrets = TempDir::new().unwrap();
    let auth_root = TempDir::new().unwrap();
    std::env::set_var("CAR_SECRETS_FILE_DIR", secrets.path());
    std::env::set_var(
        "CAR_AUTH_LOCK_PATH",
        auth_root.path().join("auth-coordinator.lock"),
    );
    let (api_base, registered) = spawn_session_and_mobile_mock();
    car_auth::commit_login(
        &api_base,
        &car_auth::TokenSet {
            access_token: "activation-access".into(),
            refresh_token: "activation-refresh".into(),
            expires_in: 3_600,
            token_type: "Bearer".into(),
        },
        r#"{"Account":{"Id":"account-1","Email":"person@example.test"}}"#,
        None,
    )
    .await
    .unwrap();

    let state = gated_state(journal.path().to_path_buf());
    state
        .install_auth_token("local-runtime-token".to_string())
        .unwrap();
    state
        .install_mobile_registration_url("wss://runtime.example.test/car".to_string())
        .unwrap();
    let addr = spawn_dispatcher(state.clone(), 1).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    authenticate_host(&mut ws).await;

    let before = car_secrets::secret_store_activity();
    let status = call(&mut ws, "activate", "auth.status", serde_json::json!({})).await;
    assert_eq!(status["result"]["authenticated"], true, "{status}");
    assert_eq!(
        car_secrets::secret_store_activity().get_attempts,
        before.get_attempts + 1,
        "explicit Verify should perform one authoritative credential read"
    );
    let active = state
        .parslee_session
        .get()
        .expect("explicit status should activate daemon identity");
    assert_eq!(active.identity.account_id, "account-1");
    assert_eq!(
        active.identity.email.as_deref(),
        Some("person@example.test")
    );
    tokio::task::spawn_blocking(move || registered.recv_timeout(EVENT_WAIT))
        .await
        .expect("join mobile registration observer")
        .expect("explicit activation should register the configured mobile runtime");
}

#[tokio::test]
async fn parslee_auth_activates_once_from_one_authoritative_read() {
    const SENTINEL: &str = "CAR_TASK3_PARSLEE_AUTH_ACTIVATION_CHILD";
    if run_isolated_contract(
        "parslee_auth_activates_once_from_one_authoritative_read",
        SENTINEL,
    ) {
        return;
    }

    let _env_guard = auth_env_lock().lock().await;
    let journal = TempDir::new().unwrap();
    let secrets = TempDir::new().unwrap();
    let auth_root = TempDir::new().unwrap();
    std::env::set_var("CAR_SECRETS_FILE_DIR", secrets.path());
    std::env::set_var(
        "CAR_AUTH_LOCK_PATH",
        auth_root.path().join("auth-coordinator.lock"),
    );
    let (api_base, registered, requests) = spawn_repeated_session_and_mobile_mock();
    assert!(
        api_base.starts_with("http://127.0.0.1:"),
        "auth test authority must be loopback"
    );
    car_auth::commit_login(
        &api_base,
        &car_auth::TokenSet {
            access_token: "parslee-auth-access".into(),
            refresh_token: "parslee-auth-refresh".into(),
            expires_in: 3_600,
            token_type: "Bearer".into(),
        },
        r#"{"Account":{"Id":"account-1","Email":"person@example.test"}}"#,
        None,
    )
    .await
    .unwrap();

    let state = gated_state(journal.path().to_path_buf());
    state
        .install_auth_token("local-runtime-token".to_string())
        .unwrap();
    state
        .install_mobile_registration_url("wss://runtime.example.test/car".to_string())
        .unwrap();
    let addr = spawn_dispatcher(state.clone(), 1).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    authenticate_host(&mut ws).await;

    let before = car_secrets::secret_store_activity();
    let first = call(
        &mut ws,
        "parslee-auth-1",
        "parslee.auth",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(first["result"]["authenticated"], true);
    assert_eq!(
        car_secrets::secret_store_activity().get_attempts,
        before.get_attempts + 1,
        "parslee.auth must resolve one authoritative credential bundle"
    );
    let active = state
        .parslee_session
        .get()
        .expect("parslee.auth should activate daemon identity");
    assert_eq!(active.identity.account_id, "account-1");
    tokio::task::spawn_blocking(move || registered.recv_timeout(EVENT_WAIT))
        .await
        .expect("join parslee.auth registration observer")
        .expect("parslee.auth should register the configured mobile runtime");

    let second = call(
        &mut ws,
        "parslee-auth-2",
        "parslee.auth",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(second["result"]["authenticated"], true);
    assert_eq!(
        car_secrets::secret_store_activity().get_attempts,
        before.get_attempts + 1,
        "the coordinator may reuse the first healthy credential generation"
    );
    let observed = tokio::task::spawn_blocking(move || requests.recv_timeout(EVENT_WAIT))
        .await
        .expect("join parslee.auth request observer")
        .expect("loopback auth mock should publish its requests");
    assert_eq!(
        observed,
        vec![
            "GET /connect/session HTTP/1.1",
            "POST /mobile/runtimes HTTP/1.1",
            "GET /connect/session HTTP/1.1",
        ],
        "repeated explicit auth must not duplicate mobile registration"
    );
}

#[tokio::test]
async fn credential_events_are_host_gated_and_generation_ordered() {
    const SENTINEL: &str = "CAR_TASK3_EVENT_CONTRACT_CHILD";
    if run_isolated_contract(
        "credential_events_are_host_gated_and_generation_ordered",
        SENTINEL,
    ) {
        return;
    }

    let _env_guard = auth_env_lock().lock().await;
    let journal = TempDir::new().unwrap();
    let secrets = TempDir::new().unwrap();
    std::env::set_var("CAR_SECRETS_FILE_DIR", secrets.path());
    let api_base = spawn_session_mock();
    car_auth::commit_login(
        &api_base,
        &car_auth::TokenSet {
            access_token: "access-1".into(),
            refresh_token: "refresh-1".into(),
            expires_in: 3_600,
            token_type: "Bearer".into(),
        },
        r#"{"Account":{"Id":"account-1","Email":"person@example.test"}}"#,
        None,
    )
    .await
    .unwrap();

    let state = gated_state(journal.path().to_path_buf());
    let addr = spawn_dispatcher(state, 2).await;
    let (mut non_host, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    negotiate(&mut non_host).await;
    let (mut host, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    authenticate_host(&mut host).await;
    tokio::task::yield_now().await;

    host.send(Message::Text(
        serde_json::json!({
            "jsonrpc": "2.0",
            "id": "status",
            "method": "auth.status",
            "params": {},
        })
        .to_string()
        .into(),
    ))
    .await
    .unwrap();

    let deadline = std::time::Instant::now() + EVENT_WAIT;
    let mut states = Vec::new();
    let mut generations = Vec::new();
    let mut saw_response = false;
    while std::time::Instant::now() < deadline && (!saw_response || states.len() < 2) {
        let frame = tokio::time::timeout(EVENT_WAIT, host.next())
            .await
            .expect("credential event or status response")
            .expect("host frame")
            .expect("host frame ok");
        let value: serde_json::Value =
            serde_json::from_str(&frame.into_text().unwrap()).expect("credential frame JSON");
        if value["id"] == "status" {
            assert_eq!(value["result"]["authenticated"], true, "{value}");
            saw_response = true;
        } else if value["method"] == "auth.credential.event" {
            states.push(value["params"]["state"].as_str().unwrap().to_string());
            generations.push(value["params"]["generation"].as_u64().unwrap());
        }
    }
    assert_eq!(states, ["pending", "configured"]);
    assert!(
        generations.windows(2).all(|pair| pair[0] <= pair[1]),
        "credential generations must be monotonic: {generations:?}"
    );
    assert!(
        tokio::time::timeout(std::time::Duration::from_millis(250), non_host.next())
            .await
            .is_err(),
        "a non-host connection must receive no credential events"
    );
}

#[tokio::test]
async fn secret_activity_diagnostics_are_host_only_and_aggregate() {
    const SENTINEL: &str = "CAR_TASK3_DIAGNOSTICS_CONTRACT_CHILD";
    if run_isolated_contract(
        "secret_activity_diagnostics_are_host_only_and_aggregate",
        SENTINEL,
    ) {
        return;
    }

    let _env_guard = auth_env_lock().lock().await;
    let journal = TempDir::new().unwrap();
    let fake_home = TempDir::new().unwrap();
    let secrets = TempDir::new().unwrap();
    std::env::set_var("HOME", fake_home.path());
    std::env::set_var("USERPROFILE", fake_home.path());
    std::env::set_var("CAR_SECRETS_FILE_DIR", secrets.path());
    std::env::set_var("CAR_NO_INFERENCE_WORKER", "1");
    for key in [
        "PARSLEE_ACCESS_TOKEN",
        "PARSLEE_API_BASE",
        "OPENROUTER_API_KEY",
        "OPENAI_API_KEY",
        "ANTHROPIC_API_KEY",
        "GOOGLE_API_KEY",
        "GEMINI_API_KEY",
        "ELEVENLABS_API_KEY",
    ] {
        std::env::remove_var(key);
    }

    let state = gated_state(journal.path().to_path_buf());
    let addr = spawn_dispatcher(state, 2).await;
    let (mut non_host, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    negotiate(&mut non_host).await;
    let rejected = call(
        &mut non_host,
        "rejected",
        "diagnostics.secret_store_activity",
        serde_json::json!({}),
    )
    .await;
    assert!(
        rejected["error"]["message"]
            .as_str()
            .is_some_and(|message| message.contains("host-management role")),
        "diagnostics must reject non-host callers: {rejected}"
    );

    let (mut host, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    authenticate_host(&mut host).await;
    for (id, method, params) in [
        ("models", "models.list_unified", serde_json::json!({})),
        (
            "setup",
            "models.setup_plan",
            serde_json::json!({ "cloud_ok": true }),
        ),
        (
            "concierge",
            "concierge.status",
            serde_json::json!({ "inference_active": false }),
        ),
        ("voice", "voice.providers.list", serde_json::json!({})),
    ] {
        let response = call(&mut host, id, method, params).await;
        assert!(
            response.get("error").is_none(),
            "passive surface {method} failed: {response}"
        );
    }

    let authority = call(
        &mut host,
        "hint",
        "auth.authority_hint",
        serde_json::json!({}),
    )
    .await;
    assert!(authority["result"]["state"].is_string(), "{authority}");
    assert!(authority["result"]["generation"].is_number(), "{authority}");

    let diagnostics = call(
        &mut host,
        "diagnostics",
        "diagnostics.secret_store_activity",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(
        diagnostics["result"],
        serde_json::json!({
            "get_attempts": 0,
            "status_attempts": 0,
            "availability_attempts": 0,
            "write_attempts": 0,
            "delete_attempts": 0,
        }),
        "the diagnostic is aggregate-only and passive surfaces remain at zero: {diagnostics}"
    );
}