fabro/lib/crates/fabro-server/tests/it/api.rs

// ===========================================================================
// mTLS end-to-end tests
// ===========================================================================

#![allow(clippy::absolute_paths)]

fn api(path: &str) -> String {
    format!("/api/v1{path}")
}

// Skip on macOS: LibreSSL generates certs with extensions rustls rejects
#[cfg(target_os = "linux")]
mod mtls_e2e {
    use super::api;
    use std::path::Path;
    use std::process::{Command, Stdio};
    use std::sync::Arc;

    use fabro_server::jwt_auth::{AuthMode, AuthStrategy};
    use fabro_server::server::{build_router, create_app_state};
    use fabro_server::server_config::TlsSettings;
    use fabro_server::tls::{ClientAuth, build_rustls_config};
    use tokio::net::TcpListener;

    async fn test_db() -> sqlx::SqlitePool {
        let pool = fabro_db::connect_memory().await.unwrap();
        fabro_db::initialize_db(&pool).await.unwrap();
        pool
    }

    /// Generate a complete CA + server cert + client cert PKI in `dir`.
    /// Returns paths: (ca_cert, server_cert, server_key, client_cert_pem, client_key_pem)
    fn generate_pki(dir: &Path, ca_cn: &str, server_cn: &str, client_cn: &str) -> PkiPaths {
        // CA key
        let ca_key_path = dir.join("ca.key");
        let ca_cert_path = dir.join("ca.crt");
        run_openssl(&[
            "genpkey",
            "-algorithm",
            "Ed25519",
            "-out",
            ca_key_path.to_str().unwrap(),
        ]);
        run_openssl(&[
            "req",
            "-new",
            "-x509",
            "-key",
            ca_key_path.to_str().unwrap(),
            "-out",
            ca_cert_path.to_str().unwrap(),
            "-days",
            "1",
            "-subj",
            &format!("/CN={ca_cn}"),
            "-addext",
            "basicConstraints=critical,CA:TRUE",
            "-addext",
            "keyUsage=critical,keyCertSign,cRLSign",
        ]);

        // Server key + cert signed by CA
        let server_key_path = dir.join("server.key");
        let server_csr_path = dir.join("server.csr");
        let server_cert_path = dir.join("server.crt");
        run_openssl(&[
            "genpkey",
            "-algorithm",
            "Ed25519",
            "-out",
            server_key_path.to_str().unwrap(),
        ]);
        run_openssl(&[
            "req",
            "-new",
            "-key",
            server_key_path.to_str().unwrap(),
            "-out",
            server_csr_path.to_str().unwrap(),
            "-subj",
            &format!("/CN={server_cn}"),
        ]);

        // Create extension file for SAN (reqwest validates server cert hostname)
        let ext_path = dir.join("server.ext");
        std::fs::write(&ext_path, "subjectAltName=IP:127.0.0.1").unwrap();

        run_openssl(&[
            "x509",
            "-req",
            "-in",
            server_csr_path.to_str().unwrap(),
            "-CA",
            ca_cert_path.to_str().unwrap(),
            "-CAkey",
            ca_key_path.to_str().unwrap(),
            "-CAcreateserial",
            "-out",
            server_cert_path.to_str().unwrap(),
            "-days",
            "1",
            "-extfile",
            ext_path.to_str().unwrap(),
        ]);

        // Client key + cert signed by CA
        let client_key_path = dir.join("client.key");
        let client_csr_path = dir.join("client.csr");
        let client_cert_path = dir.join("client.crt");
        run_openssl(&[
            "genpkey",
            "-algorithm",
            "Ed25519",
            "-out",
            client_key_path.to_str().unwrap(),
        ]);
        run_openssl(&[
            "req",
            "-new",
            "-key",
            client_key_path.to_str().unwrap(),
            "-out",
            client_csr_path.to_str().unwrap(),
            "-subj",
            &format!("/CN={client_cn}"),
        ]);
        // Client extension file to produce a v3 certificate
        let client_ext_path = dir.join("client.ext");
        std::fs::write(&client_ext_path, "basicConstraints=CA:FALSE\n").unwrap();
        run_openssl(&[
            "x509",
            "-req",
            "-in",
            client_csr_path.to_str().unwrap(),
            "-CA",
            ca_cert_path.to_str().unwrap(),
            "-CAkey",
            ca_key_path.to_str().unwrap(),
            "-CAcreateserial",
            "-out",
            client_cert_path.to_str().unwrap(),
            "-days",
            "1",
            "-extfile",
            client_ext_path.to_str().unwrap(),
        ]);

        PkiPaths {
            ca_cert: ca_cert_path,
            server_cert: server_cert_path,
            server_key: server_key_path,
            client_cert: client_cert_path,
            client_key: client_key_path,
        }
    }

    struct PkiPaths {
        ca_cert: std::path::PathBuf,
        server_cert: std::path::PathBuf,
        server_key: std::path::PathBuf,
        client_cert: std::path::PathBuf,
        client_key: std::path::PathBuf,
    }

    fn run_openssl(args: &[&str]) {
        let output = Command::new("openssl")
            .args(args)
            .stdin(Stdio::null())
            .output()
            .expect("openssl command failed to execute");
        assert!(
            output.status.success(),
            "openssl {} failed: {}",
            args[0],
            String::from_utf8_lossy(&output.stderr)
        );
    }

    /// Start a TLS server on a random port, returning the bound address.
    async fn start_tls_server(
        tls_settings: &TlsSettings,
        client_auth: ClientAuth,
        auth_mode: AuthMode,
    ) -> std::net::SocketAddr {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        let rustls_config = build_rustls_config(tls_settings, client_auth);
        let tls_acceptor = tokio_rustls::TlsAcceptor::from(rustls_config);

        let state = create_app_state(test_db().await);
        let router = build_router(state, auth_mode);

        tokio::spawn(async move {
            let _ = fabro_server::tls::serve_tls(listener, tls_acceptor, router).await;
        });

        addr
    }

    /// Build a reqwest client with the given CA cert and optional client identity.
    fn build_client(
        ca_cert_path: &Path,
        client_cert_path: Option<&Path>,
        client_key_path: Option<&Path>,
    ) -> reqwest::Client {
        let ca_pem = std::fs::read(ca_cert_path).unwrap();
        let ca_cert = reqwest::tls::Certificate::from_pem(&ca_pem).unwrap();

        let mut builder = reqwest::Client::builder()
            .add_root_certificate(ca_cert)
            .use_rustls_tls();

        if let (Some(cert_path), Some(key_path)) = (client_cert_path, client_key_path) {
            let cert_pem = std::fs::read(cert_path).unwrap();
            let key_pem = std::fs::read(key_path).unwrap();
            let mut identity_pem = cert_pem;
            identity_pem.extend_from_slice(&key_pem);
            let identity = reqwest::tls::Identity::from_pem(&identity_pem).unwrap();
            builder = builder.identity(identity);
        }

        builder.build().unwrap()
    }

    fn install_crypto_provider() {
        let _ = rustls::crypto::ring::default_provider().install_default();
    }

    #[tokio::test]
    async fn mtls_accepts_valid_client_cert() {
        install_crypto_provider();
        let dir = tempfile::tempdir().unwrap();
        let pki = generate_pki(dir.path(), "TestCA", "localhost", "testuser");

        let tls_settings = TlsSettings {
            cert: pki.server_cert.clone(),
            key: pki.server_key.clone(),
            ca: pki.ca_cert.clone(),
        };

        let auth_mode = AuthMode::Strategies(vec![AuthStrategy::Mtls]);
        let addr = start_tls_server(&tls_settings, ClientAuth::Required, auth_mode).await;

        let client = build_client(&pki.ca_cert, Some(&pki.client_cert), Some(&pki.client_key));

        let response = client
            .get(format!("https://127.0.0.1:{}{}", addr.port(), api("/runs")))
            .send()
            .await
            .expect("request with valid client cert should succeed");

        assert_eq!(response.status(), 200);
    }

    #[tokio::test]
    async fn mtls_only_rejects_wrong_ca_client_cert() {
        install_crypto_provider();
        let dir = tempfile::tempdir().unwrap();
        let pki = generate_pki(dir.path(), "TestCA", "localhost", "testuser");

        let tls_settings = TlsSettings {
            cert: pki.server_cert.clone(),
            key: pki.server_key.clone(),
            ca: pki.ca_cert.clone(),
        };

        let auth_mode = AuthMode::Strategies(vec![AuthStrategy::Mtls]);
        let addr = start_tls_server(&tls_settings, ClientAuth::Required, auth_mode).await;

        // Generate a DIFFERENT CA and client cert signed by it
        let wrong_dir = dir.path().join("wrong_ca");
        std::fs::create_dir_all(&wrong_dir).unwrap();
        let wrong_pki = generate_pki(&wrong_dir, "WrongCA", "localhost", "intruder");

        // Client trusts the REAL server CA, but presents a cert from the WRONG CA
        let client = build_client(
            &pki.ca_cert,
            Some(&wrong_pki.client_cert),
            Some(&wrong_pki.client_key),
        );

        let result = client
            .get(format!("https://127.0.0.1:{}{}", addr.port(), api("/runs")))
            .send()
            .await;

        // Server should reject the TLS handshake — the wrong CA client cert
        // will cause a connection error (not an HTTP error)
        assert!(
            result.is_err(),
            "request with wrong-CA client cert should fail at TLS level, but got: {:?}",
            result.unwrap().status()
        );
    }

    #[tokio::test]
    async fn mtls_only_rejects_no_client_cert() {
        install_crypto_provider();
        let dir = tempfile::tempdir().unwrap();
        let pki = generate_pki(dir.path(), "TestCA", "localhost", "testuser");

        let tls_settings = TlsSettings {
            cert: pki.server_cert.clone(),
            key: pki.server_key.clone(),
            ca: pki.ca_cert.clone(),
        };

        // mTLS is the ONLY strategy → client cert is required at TLS level
        let auth_mode = AuthMode::Strategies(vec![AuthStrategy::Mtls]);
        let addr = start_tls_server(&tls_settings, ClientAuth::Required, auth_mode).await;

        // Client trusts the server CA but presents NO client cert
        let client = build_client(&pki.ca_cert, None, None);

        let result = client
            .get(format!("https://127.0.0.1:{}{}", addr.port(), api("/runs")))
            .send()
            .await;

        // Server requires client cert → TLS handshake should fail
        assert!(
            result.is_err(),
            "request without client cert should fail when mTLS is the only strategy, but got: {:?}",
            result.unwrap().status()
        );
    }

    /// Generate an Ed25519 JWT keypair. Returns (encoding_key, decoding_key).
    fn generate_jwt_keypair() -> (jsonwebtoken::EncodingKey, jsonwebtoken::DecodingKey) {
        let output = Command::new("openssl")
            .args(["genpkey", "-algorithm", "Ed25519"])
            .output()
            .expect("openssl must be available for tests");
        let private_pem = output.stdout;

        let output = Command::new("openssl")
            .args(["pkey", "-pubout"])
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .spawn()
            .and_then(|mut child| {
                use std::io::Write;
                child.stdin.take().unwrap().write_all(&private_pem).unwrap();
                child.wait_with_output()
            })
            .expect("openssl pkey failed");
        let public_pem = output.stdout;

        let encoding =
            jsonwebtoken::EncodingKey::from_ed_pem(&private_pem).expect("invalid private key");
        let decoding =
            jsonwebtoken::DecodingKey::from_ed_pem(&public_pem).expect("invalid public key");
        (encoding, decoding)
    }

    fn sign_jwt(key: &jsonwebtoken::EncodingKey, sub: &str) -> String {
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_secs();
        let claims = serde_json::json!({
            "iss": "fabro-web",
            "iat": now,
            "exp": now + 60,
            "sub": sub,
        });
        let header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::EdDSA);
        jsonwebtoken::encode(&header, &claims, key).expect("failed to sign token")
    }

    #[tokio::test]
    async fn mtls_and_jwt_accepts_valid_jwt_without_client_cert() {
        install_crypto_provider();
        let dir = tempfile::tempdir().unwrap();
        let pki = generate_pki(dir.path(), "TestCA", "localhost", "testuser");

        let tls_settings = TlsSettings {
            cert: pki.server_cert.clone(),
            key: pki.server_key.clone(),
            ca: pki.ca_cert.clone(),
        };

        let (encoding_key, decoding_key) = generate_jwt_keypair();

        // Both mTLS and JWT strategies; mTLS is optional since JWT is also present
        let auth_mode = AuthMode::Strategies(vec![
            AuthStrategy::Mtls,
            AuthStrategy::Jwt {
                key: Arc::new(decoding_key),
                validation: Arc::new(fabro_server::jwt_auth::jwt_validation()),
                allowed_usernames: vec!["brynary".to_string()],
            },
        ]);
        let addr = start_tls_server(&tls_settings, ClientAuth::Optional, auth_mode).await;

        // Client trusts the server CA but presents NO client cert
        let client = build_client(&pki.ca_cert, None, None);

        let token = sign_jwt(&encoding_key, "https://github.com/brynary");

        let response = client
            .get(format!("https://127.0.0.1:{}{}", addr.port(), api("/runs")))
            .bearer_auth(&token)
            .send()
            .await
            .expect("request with valid JWT and no client cert should succeed");

        assert_eq!(
            response.status(),
            200,
            "valid JWT should be accepted when strategies = [mtls, jwt]"
        );
    }
}

// ===========================================================================
// Full HTTP server lifecycle (TS Scenario 4)
// ===========================================================================

mod server_lifecycle {
    use super::super::helpers::test_db;
    use super::api;
    use std::sync::Arc;
    use std::time::Duration;

    use axum::body::Body;
    use axum::http::{Request, StatusCode};
    use fabro_interview::Interviewer;
    use fabro_server::server::{build_router, create_app_state_with_registry_factory};
    use fabro_workflow::handler::HandlerRegistry;
    use fabro_workflow::handler::agent::AgentHandler;
    use fabro_workflow::handler::exit::ExitHandler;
    use fabro_workflow::handler::human::HumanHandler;
    use fabro_workflow::handler::start::StartHandler;
    use tower::ServiceExt;

    fn gate_registry(interviewer: Arc<dyn Interviewer>) -> HandlerRegistry {
        let mut registry = HandlerRegistry::new(Box::new(AgentHandler::new(None)));
        registry.register("start", Box::new(StartHandler));
        registry.register("exit", Box::new(ExitHandler));
        registry.register("agent", Box::new(AgentHandler::new(None)));
        registry.register("human", Box::new(HumanHandler::new(interviewer)));
        registry
    }

    async fn body_json(body: Body) -> serde_json::Value {
        let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap();
        serde_json::from_slice(&bytes).unwrap()
    }

    const POLL_INTERVAL: Duration = Duration::from_millis(10);
    const POLL_ATTEMPTS: usize = 500;

    async fn run_json(app: &axum::Router, run_id: &str) -> serde_json::Value {
        let req = Request::builder()
            .method("GET")
            .uri(api(&format!("/runs/{run_id}")))
            .body(Body::empty())
            .unwrap();
        let response = app.clone().oneshot(req).await.unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        body_json(response.into_body()).await
    }

    async fn wait_for_question_id(app: &axum::Router, run_id: &str) -> String {
        for _ in 0..POLL_ATTEMPTS {
            let req = Request::builder()
                .method("GET")
                .uri(api(&format!("/runs/{run_id}/questions")))
                .body(Body::empty())
                .unwrap();
            let response = app.clone().oneshot(req).await.unwrap();
            let body = body_json(response.into_body()).await;
            let arr = body["data"].as_array().unwrap();
            if let Some(question_id) = arr
                .first()
                .and_then(|item| item["id"].as_str())
                .map(ToOwned::to_owned)
            {
                return question_id;
            }
            tokio::time::sleep(POLL_INTERVAL).await;
        }
        panic!("question should have appeared");
    }

    async fn wait_for_run_status(app: &axum::Router, run_id: &str, expected: &[&str]) -> String {
        for _ in 0..POLL_ATTEMPTS {
            let body = run_json(app, run_id).await;
            let status = body["status"].as_str().unwrap().to_string();
            if expected.iter().any(|candidate| *candidate == status) {
                return status;
            }
            tokio::time::sleep(POLL_INTERVAL).await;
        }
        panic!("run {run_id} did not reach any of {expected:?}");
    }

    async fn wait_for_run_status_not_in(
        app: &axum::Router,
        run_id: &str,
        unexpected: &[&str],
    ) -> String {
        for _ in 0..POLL_ATTEMPTS {
            let body = run_json(app, run_id).await;
            let status = body["status"].as_str().unwrap().to_string();
            if unexpected.iter().all(|candidate| *candidate != status) {
                return status;
            }
            tokio::time::sleep(POLL_INTERVAL).await;
        }
        panic!("run {run_id} stayed in {unexpected:?}");
    }

    const GATE_DOT: &str = r#"digraph GateTest {
        graph [goal="Test gate"]
        start [shape=Mdiamond]
        exit  [shape=Msquare]
        work  [shape=box, prompt="Do work"]
        gate  [shape=hexagon, type="human", label="Approve?"]
        done  [shape=box, prompt="Finish"]
        revise [shape=box, prompt="Revise"]

        start -> work -> gate
        gate -> done   [label="[A] Approve"]
        gate -> revise [label="[R] Revise"]
        done -> exit
        revise -> gate
    }"#;

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn full_http_lifecycle_approve_and_complete() {
        let state = create_app_state_with_registry_factory(test_db().await, gate_registry);
        fabro_server::server::spawn_scheduler(Arc::clone(&state));
        let app = build_router(
            Arc::clone(&state),
            fabro_server::jwt_auth::AuthMode::Disabled,
        );

        // 1. Start run
        let req = Request::builder()
            .method("POST")
            .uri(api("/runs"))
            .header("content-type", "application/json")
            .body(Body::from(
                serde_json::to_string(&serde_json::json!({"dot_source": GATE_DOT})).unwrap(),
            ))
            .unwrap();

        let response = app.clone().oneshot(req).await.unwrap();
        assert_eq!(response.status(), StatusCode::CREATED);
        let body = body_json(response.into_body()).await;
        let run_id = body["id"].as_str().unwrap().to_string();

        // 2. Poll for question to appear (run goes start -> work -> gate, then blocks)
        let question_id = wait_for_question_id(&app, &run_id).await;

        // 3. Submit answer selecting first option (Approve)
        let req = Request::builder()
            .method("POST")
            .uri(api(&format!(
                "/runs/{run_id}/questions/{question_id}/answer"
            )))
            .header("content-type", "application/json")
            .body(Body::from(
                serde_json::to_string(&serde_json::json!({"value": "A"})).unwrap(),
            ))
            .unwrap();
        let response = app.clone().oneshot(req).await.unwrap();
        assert_eq!(response.status(), StatusCode::NO_CONTENT);

        // 4. Poll until completed
        let final_status = wait_for_run_status(&app, &run_id, &["completed", "failed"]).await;
        assert_eq!(final_status, "completed");

        // 5. Verify context endpoint returns an object
        let req = Request::builder()
            .method("GET")
            .uri(api(&format!("/runs/{run_id}/context")))
            .body(Body::empty())
            .unwrap();
        let response = app.clone().oneshot(req).await.unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        let ctx_body = body_json(response.into_body()).await;
        assert!(ctx_body.is_object(), "context should be an object");

        // 6. Verify no pending questions
        let req = Request::builder()
            .method("GET")
            .uri(api(&format!("/runs/{run_id}/questions")))
            .body(Body::empty())
            .unwrap();
        let response = app.clone().oneshot(req).await.unwrap();
        let body = body_json(response.into_body()).await;
        assert!(
            body["data"].as_array().unwrap().is_empty(),
            "no pending questions after completion"
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn full_http_lifecycle_cancel() {
        let state = create_app_state_with_registry_factory(test_db().await, gate_registry);
        fabro_server::server::spawn_scheduler(Arc::clone(&state));
        let app = build_router(
            Arc::clone(&state),
            fabro_server::jwt_auth::AuthMode::Disabled,
        );

        // Start a run that will block at the human gate
        let req = Request::builder()
            .method("POST")
            .uri(api("/runs"))
            .header("content-type", "application/json")
            .body(Body::from(
                serde_json::to_string(&serde_json::json!({"dot_source": GATE_DOT})).unwrap(),
            ))
            .unwrap();
        let response = app.clone().oneshot(req).await.unwrap();
        let body = body_json(response.into_body()).await;
        let run_id = body["id"].as_str().unwrap().to_string();

        // Subscribe as soon as the scheduler has created the live event stream.
        // Waiting past "starting" races with stage events because `/events`
        // subscribes to future broadcast messages only; it does not replay.
        wait_for_run_status_not_in(&app, &run_id, &["queued"]).await;

        // Cancel it
        let req = Request::builder()
            .method("POST")
            .uri(api(&format!("/runs/{run_id}/cancel")))
            .body(Body::empty())
            .unwrap();
        let response = app.clone().oneshot(req).await.unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        let body = body_json(response.into_body()).await;
        assert_eq!(body["status"], "cancelled");

        // Verify status is cancelled
        let req = Request::builder()
            .method("GET")
            .uri(api(&format!("/runs/{run_id}")))
            .body(Body::empty())
            .unwrap();
        let response = app.clone().oneshot(req).await.unwrap();
        let body = body_json(response.into_body()).await;
        assert_eq!(body["status"], "cancelled");
    }
}

// ===========================================================================
// SSE event stream content parsing (TS Scenario 8)
// ===========================================================================

mod sse_events {
    use super::super::helpers::test_db;
    use super::api;
    use std::sync::Arc;
    use std::time::Duration;

    use axum::body::Body;
    use axum::http::{Request, StatusCode};
    use fabro_server::server::{build_router, create_app_state_with_options};
    use fabro_types::settings::FabroSettings;
    use http_body_util::BodyExt;
    use tower::ServiceExt;

    const SIMPLE_DOT: &str = r#"digraph SSETest {
        graph [goal="Test SSE"]
        start [shape=Mdiamond]
        work  [shape=box, prompt="Do work"]
        exit  [shape=Msquare]
        start -> work -> exit
    }"#;

    const POLL_INTERVAL: Duration = Duration::from_millis(10);
    const POLL_ATTEMPTS: usize = 500;

    fn dry_run_settings() -> FabroSettings {
        FabroSettings {
            dry_run: Some(true),
            ..Default::default()
        }
    }

    async fn body_json(body: Body) -> serde_json::Value {
        let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap();
        serde_json::from_slice(&bytes).unwrap()
    }

    async fn run_status(app: &axum::Router, run_id: &str) -> String {
        let req = Request::builder()
            .method("GET")
            .uri(api(&format!("/runs/{run_id}")))
            .body(Body::empty())
            .unwrap();
        let response = app.clone().oneshot(req).await.unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        let body = body_json(response.into_body()).await;
        body["status"].as_str().unwrap().to_string()
    }

    async fn wait_for_run_status_not_in(
        app: &axum::Router,
        run_id: &str,
        unexpected: &[&str],
    ) -> String {
        for _ in 0..POLL_ATTEMPTS {
            let status = run_status(app, run_id).await;
            if unexpected.iter().all(|candidate| *candidate != status) {
                return status;
            }
            tokio::time::sleep(POLL_INTERVAL).await;
        }
        panic!("run {run_id} stayed in {unexpected:?}");
    }

    async fn wait_for_checkpoint(app: &axum::Router, run_id: &str) -> serde_json::Value {
        for _ in 0..POLL_ATTEMPTS {
            let req = Request::builder()
                .method("GET")
                .uri(api(&format!("/runs/{run_id}/checkpoint")))
                .body(Body::empty())
                .unwrap();
            let response = app.clone().oneshot(req).await.unwrap();
            if response.status() == StatusCode::OK {
                return body_json(response.into_body()).await;
            }
            tokio::time::sleep(POLL_INTERVAL).await;
        }
        panic!("checkpoint did not become available for {run_id}");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn sse_stream_contains_expected_event_types() {
        let state = create_app_state_with_options(test_db().await, dry_run_settings(), 5);
        fabro_server::server::spawn_scheduler(Arc::clone(&state));
        let app = build_router(
            Arc::clone(&state),
            fabro_server::jwt_auth::AuthMode::Disabled,
        );

        // Start run
        let req = Request::builder()
            .method("POST")
            .uri(api("/runs"))
            .header("content-type", "application/json")
            .body(Body::from(
                serde_json::to_string(&serde_json::json!({"dot_source": SIMPLE_DOT})).unwrap(),
            ))
            .unwrap();
        let response = app.clone().oneshot(req).await.unwrap();
        assert_eq!(response.status(), StatusCode::CREATED);
        let body = body_json(response.into_body()).await;
        let run_id = body["id"].as_str().unwrap().to_string();

        wait_for_run_status_not_in(&app, &run_id, &["queued", "starting"]).await;

        // Get SSE stream
        let req = Request::builder()
            .method("GET")
            .uri(api(&format!("/runs/{run_id}/events")))
            .body(Body::empty())
            .unwrap();
        let response = app.clone().oneshot(req).await.unwrap();
        // May be 200 (stream open) or 410 (run completed before connect)
        let sse_status = response.status();
        assert!(
            sse_status == StatusCode::OK || sse_status == StatusCode::GONE,
            "expected 200 or 410, got: {sse_status}"
        );
        if sse_status == StatusCode::GONE {
            return;
        }

        let content_type = response
            .headers()
            .get("content-type")
            .unwrap()
            .to_str()
            .unwrap();
        assert!(content_type.contains("text/event-stream"));

        // Collect SSE frames with a timeout
        let mut body = response.into_body();
        let mut sse_data = String::new();
        while let Ok(Some(Ok(frame))) =
            tokio::time::timeout(Duration::from_millis(500), body.frame()).await
        {
            if let Some(data) = frame.data_ref() {
                sse_data.push_str(&String::from_utf8_lossy(data));
            }
        }

        // Parse SSE data lines and extract event types
        let mut event_types: Vec<String> = Vec::new();
        for line in sse_data.lines() {
            if let Some(json_str) = line.strip_prefix("data:") {
                let json_str = json_str.trim();
                if let Ok(event) = serde_json::from_str::<serde_json::Value>(json_str) {
                    if let Some(event_name) = event["event"].as_str() {
                        event_types.push(event_name.to_string());
                    }
                }
            }
        }

        // Because we subscribe while the run is only guaranteed to be past
        // "queued", a live stream should include at least one stage event.
        // A 410 response above still covers the case where the run completed
        // before we managed to attach.
        if !event_types.is_empty() {
            assert!(
                event_types
                    .iter()
                    .any(|t| t == "stage.started" || t == "stage.completed"),
                "should contain stage events, got: {event_types:?}"
            );
        }

        // Pipeline is complete (SSE stream ended), verify checkpoint
        let cp_body = wait_for_checkpoint(&app, &run_id).await;
        // If run completed, checkpoint should have completed_nodes
        if !cp_body.is_null() {
            let completed = cp_body["completed_nodes"].as_array();
            if let Some(nodes) = completed {
                let names: Vec<&str> = nodes.iter().filter_map(|v| v.as_str()).collect();
                assert!(names.contains(&"work"), "work should be in completed_nodes");
            }
        }
    }
}

// ===========================================================================
// Serve command: dry-run registry factory builds a working router
// ===========================================================================

mod serve_dry_run {
    use super::super::helpers::test_db;
    use super::api;
    use std::sync::Arc;
    use std::time::Duration;

    use axum::body::Body;
    use axum::http::{Request, StatusCode};
    use fabro_server::server::{build_router, create_app_state_with_options};
    use tower::ServiceExt;

    const MINIMAL_DOT: &str = r#"digraph Test {
        graph [goal="Test"]
        start [shape=Mdiamond]
        exit  [shape=Msquare]
        start -> exit
    }"#;

    /// Build the router exactly as `serve_command` does in dry-run mode.
    async fn dry_run_app() -> axum::Router {
        let state = create_app_state_with_options(
            test_db().await,
            fabro_config::FabroSettings {
                dry_run: Some(true),
                ..Default::default()
            },
            5,
        );
        fabro_server::server::spawn_scheduler(Arc::clone(&state));
        build_router(state, fabro_server::jwt_auth::AuthMode::Disabled)
    }

    async fn body_json(body: Body) -> serde_json::Value {
        let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap();
        serde_json::from_slice(&bytes).unwrap()
    }

    const POLL_INTERVAL: Duration = Duration::from_millis(10);
    const POLL_ATTEMPTS: usize = 500;

    async fn wait_for_run_status(app: &axum::Router, run_id: &str, expected: &[&str]) -> String {
        for _ in 0..POLL_ATTEMPTS {
            let req = Request::builder()
                .method("GET")
                .uri(api(&format!("/runs/{run_id}")))
                .body(Body::empty())
                .unwrap();

            let response = app.clone().oneshot(req).await.unwrap();
            assert_eq!(response.status(), StatusCode::OK);
            let body = body_json(response.into_body()).await;
            let status = body["status"].as_str().unwrap().to_string();
            if expected.iter().any(|candidate| *candidate == status) {
                return status;
            }
            tokio::time::sleep(POLL_INTERVAL).await;
        }
        panic!("run {run_id} did not reach any of {expected:?}");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn dry_run_serve_starts_and_runs_workflow() {
        let app = dry_run_app().await;

        // POST /runs to start a run
        let req = Request::builder()
            .method("POST")
            .uri(api("/runs"))
            .header("content-type", "application/json")
            .body(Body::from(
                serde_json::to_string(&serde_json::json!({"dot_source": MINIMAL_DOT})).unwrap(),
            ))
            .unwrap();

        let response = app.clone().oneshot(req).await.unwrap();
        assert_eq!(response.status(), StatusCode::CREATED);

        let body = body_json(response.into_body()).await;
        let run_id = body["id"].as_str().unwrap().to_string();
        assert!(!run_id.is_empty());

        let status = wait_for_run_status(&app, &run_id, &["completed", "failed"]).await;
        assert_eq!(status, "completed");
    }

    #[tokio::test]
    async fn test_model_known_via_full_router() {
        let app = dry_run_app().await;

        let req = Request::builder()
            .method("POST")
            .uri(api("/models/claude-opus-4-6/test"))
            .header("content-type", "application/json")
            .body(Body::empty())
            .unwrap();

        let response = app.oneshot(req).await.unwrap();
        assert_eq!(response.status(), StatusCode::OK);

        let body = body_json(response.into_body()).await;
        assert_eq!(body["model_id"], "claude-opus-4-6");
        // No API keys in test env, so status will be "error"
        assert!(body["status"] == "ok" || body["status"] == "error");
    }

    #[tokio::test]
    async fn test_model_unknown_via_full_router() {
        let app = dry_run_app().await;

        let req = Request::builder()
            .method("POST")
            .uri(api("/models/nonexistent-model-xyz/test"))
            .header("content-type", "application/json")
            .body(Body::empty())
            .unwrap();

        let response = app.oneshot(req).await.unwrap();
        assert_eq!(response.status(), StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn dry_run_serve_rejects_invalid_dot() {
        let app = dry_run_app().await;

        let req = Request::builder()
            .method("POST")
            .uri(api("/runs"))
            .header("content-type", "application/json")
            .body(Body::from(
                serde_json::to_string(&serde_json::json!({"dot_source": "not valid dot"})).unwrap(),
            ))
            .unwrap();

        let response = app.oneshot(req).await.unwrap();
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
    }
}

mod route_prefixes {
    use super::super::helpers::test_db;
    use axum::body::Body;
    use axum::http::{Method, Request, StatusCode};
    use fabro_server::server::{build_router, create_app_state};
    use tower::ServiceExt;

    async fn body_json(body: Body) -> serde_json::Value {
        let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap();
        serde_json::from_slice(&bytes).unwrap()
    }

    #[tokio::test]
    async fn old_unversioned_routes_return_404() {
        let app = build_router(
            create_app_state(test_db().await),
            fabro_server::jwt_auth::AuthMode::Disabled,
        );

        let cases = [
            (Method::POST, "/completions"),
        ];

        for (method, path) in cases {
            let req = Request::builder()
                .method(method.clone())
                .uri(path)
                .body(Body::empty())
                .unwrap();
            let response = app.clone().oneshot(req).await.unwrap();
            assert_eq!(response.status(), StatusCode::NOT_FOUND, "{method} {path}");
        }
    }

    #[tokio::test]
    async fn root_and_health_stay_at_root() {
        let app = build_router(
            create_app_state(test_db().await),
            fabro_server::jwt_auth::AuthMode::Disabled,
        );

        let root_req = Request::builder()
            .method("GET")
            .uri("/")
            .body(Body::empty())
            .unwrap();
        let root_response = app.clone().oneshot(root_req).await.unwrap();
        assert_eq!(root_response.status(), StatusCode::OK);
        let root_body = axum::body::to_bytes(root_response.into_body(), usize::MAX)
            .await
            .unwrap();
        let root_html = String::from_utf8(root_body.to_vec()).unwrap();
        assert!(root_html.contains("<div id=\"root\"></div>"));

        let health_req = Request::builder()
            .method("GET")
            .uri("/health")
            .body(Body::empty())
            .unwrap();
        let health_response = app.oneshot(health_req).await.unwrap();
        assert_eq!(health_response.status(), StatusCode::OK);
        let health_body = body_json(health_response.into_body()).await;
        assert_eq!(health_body["status"], "ok");
    }

    #[tokio::test]
    async fn moved_routes_not_at_root_of_api_prefix() {
        let app = build_router(
            create_app_state(test_db().await),
            fabro_server::jwt_auth::AuthMode::Disabled,
        );

        for path in ["/api/v1/health", "/api/v1/"] {
            let req = Request::builder()
                .method("GET")
                .uri(path)
                .body(Body::empty())
                .unwrap();
            let response = app.clone().oneshot(req).await.unwrap();
            assert_eq!(response.status(), StatusCode::NOT_FOUND, "GET {path}");
        }
    }
}