car-server-core 0.47.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::{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";

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");
    let text = ws
        .next()
        .await
        .expect("frame")
        .expect("frame ok")
        .into_text()
        .expect("text")
        .to_string();
    serde_json::from_str(&text).expect("parse")
}

async fn wait_for_terminal_proof(ws: &mut Ws, attempt_id: &str) -> serde_json::Value {
    for poll in 0..100 {
        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();
        }
        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) {
    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;
        }
    }
}

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,
    }
}

fn spawn_rejecting_oauth_mock() -> (String, Arc<AtomicUsize>, thread::JoinHandle<()>) {
    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 handle = thread::spawn(move || {
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(1);
        while std::time::Instant::now() < deadline {
            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();
                }
                Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                    thread::sleep(std::time::Duration::from_millis(5));
                }
                Err(error) => panic!("accept OAuth request: {error}"),
            }
        }
    });
    (base, token_requests, 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());
    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.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(std::time::Duration::from_secs(2))
            .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(std::time::Duration::from_secs(2))
            .expect("worker should exchange the code");
    })
    .await
    .unwrap();

    let accepted = tokio::time::timeout(std::time::Duration::from_millis(250), 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 (api_base, token_requests, 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:?}"
    );

    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(std::time::Duration::from_secs(2))
            .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(std::time::Duration::from_secs(2), 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");
}