use std::{fs, time::Duration};

use axum::{
    Json, Router,
    body::Body,
    http::{HeaderMap, Method, Request, StatusCode, header},
    response::Response,
    routing::{get, post},
};
use http_body_util::BodyExt;
use serde_json::{Value, json};
use sqlx::{
    SqlitePool,
    sqlite::{SqliteConnectOptions, SqlitePoolOptions},
};
use task_pilot_backend::database;
use task_pilot_backend::state::AppState;
use tempfile::TempDir;
use tokio::time::sleep;
use tokio::{net::TcpListener, task::JoinHandle};
use tower::ServiceExt;

use task_pilot_backend::runs::executor::ExecutionEvent;

struct TestApp {
    router: Router,
    state: AppState,
    _temp_dir: TempDir,
}

async fn test_app() -> TestApp {
    let temp_dir = TempDir::new().expect("temporary directory should be created");
    let pool = task_pilot_backend::database::connect("sqlite::memory:")
        .await
        .expect("test database should initialize");
    let state = AppState::start(pool, temp_dir.path().join("logs"))
        .await
        .expect("app state should start");
    let router = task_pilot_backend::router(state.clone());
    TestApp {
        router,
        state,
        _temp_dir: temp_dir,
    }
}

async fn send_json(router: &Router, method: Method, uri: &str, body: Option<Value>) -> Response {
    let token = if matches!(method, Method::GET | Method::HEAD | Method::OPTIONS) {
        None
    } else {
        Some(fetch_api_token(router).await)
    };
    let body = body
        .map(|value| Body::from(serde_json::to_vec(&value).expect("JSON should serialize")))
        .unwrap_or_else(Body::empty);
    let mut builder = Request::builder()
        .method(method)
        .uri(uri)
        .header(header::CONTENT_TYPE, "application/json");
    if let Some(token) = token {
        builder = builder.header("x-taskpilot-token", token);
    }
    let request = builder.body(body).expect("request should build");

    router
        .clone()
        .oneshot(request)
        .await
        .expect("router should respond")
}

async fn fetch_api_token(router: &Router) -> String {
    let request = Request::builder()
        .method(Method::GET)
        .uri("/api/v1/session")
        .body(Body::empty())
        .expect("session request should build");
    let response = router
        .clone()
        .oneshot(request)
        .await
        .expect("session endpoint should respond");
    assert_eq!(response.status(), StatusCode::OK);
    response_json(response).await["token"]
        .as_str()
        .expect("session token should be a string")
        .to_owned()
}

async fn response_json(response: Response) -> Value {
    let body = response
        .into_body()
        .collect()
        .await
        .expect("response body should be readable")
        .to_bytes();
    serde_json::from_slice(&body).expect("response should be JSON")
}

async fn connect_file(path: &std::path::Path) -> SqlitePool {
    let options = SqliteConnectOptions::new()
        .filename(path)
        .create_if_missing(true)
        .foreign_keys(true);
    let pool = SqlitePoolOptions::new()
        .max_connections(1)
        .connect_with(options)
        .await
        .expect("file database should open");
    database::migrate(&pool)
        .await
        .expect("migrations should run");
    pool
}

fn script_task(
    id: &str,
    working_dir: &str,
    interpreter: &str,
    script: &str,
    args: Vec<&str>,
    cron: &str,
    timeout_secs: u64,
) -> Value {
    json!({
        "id": id,
        "name": id,
        "cron": cron,
        "timeout_secs": timeout_secs,
        "action": {
            "type": "script",
            "working_dir": working_dir,
            "interpreter": interpreter,
            "script": script,
            "args": args,
            "env": {}
        }
    })
}

fn http_task(url: &str) -> Value {
    json!({
        "id": "http_task",
        "name": "HTTP task",
        "cron": "0 0 2 * * *",
        "timeout_secs": 5,
        "action": {
            "type": "http",
            "url": url,
            "method": "POST",
            "headers": {
                "X-Test": "yes"
            },
            "body": "{\"ping\":true}",
            "expected_status": [202]
        }
    })
}

async fn spawn_http_target() -> (String, JoinHandle<()>) {
    async fn check(headers: HeaderMap, body: String) -> StatusCode {
        if headers.get("x-test").and_then(|value| value.to_str().ok()) == Some("yes")
            && body.contains("\"ping\":true")
        {
            StatusCode::ACCEPTED
        } else {
            StatusCode::BAD_REQUEST
        }
    }

    let app = Router::new()
        .route("/check", post(check))
        .route("/health", get(|| async { Json(json!({"status":"ok"})) }));
    let listener = TcpListener::bind(("127.0.0.1", 0))
        .await
        .expect("HTTP test listener should bind");
    let address = listener
        .local_addr()
        .expect("listener address should exist");
    let handle = tokio::spawn(async move {
        let _ = axum::serve(listener, app).await;
    });
    (format!("http://{address}"), handle)
}

async fn wait_for_run(router: &Router, run_id: i64) -> Value {
    for _ in 0..100 {
        let response =
            send_json(router, Method::GET, &format!("/api/v1/runs/{run_id}"), None).await;
        assert_eq!(response.status(), StatusCode::OK);
        let run = response_json(response).await;
        if matches!(
            run["status"].as_str(),
            Some("succeeded" | "failed" | "timed_out" | "cancelled" | "skipped")
        ) {
            return run;
        }
        sleep(Duration::from_millis(100)).await;
    }

    panic!("run {run_id} did not reach a terminal state");
}

#[tokio::test]
async fn shell_language_runs_batch_file_with_cmd_c() {
    let app = test_app().await;
    let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
    fs::write(
        app._temp_dir.path().join("hello.bat"),
        "@echo off\r\necho shell-language-ok\r\n",
    )
    .expect("batch file should be written");

    let mut task = script_task(
        "shell_task",
        &working_dir,
        "cmd.exe",
        "hello.bat",
        vec![],
        "0 0 2 * * *",
        10,
    );
    task["action"]["language"] = json!("shell");
    let create_response = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
    assert_eq!(create_response.status(), StatusCode::CREATED);

    let run_response = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks/shell_task/run",
        None,
    )
    .await;
    assert_eq!(run_response.status(), StatusCode::ACCEPTED);
    let run = response_json(run_response).await;
    let run_id = run["id"].as_i64().expect("run id should be an integer");

    let completed = wait_for_run(&app.router, run_id).await;
    assert_eq!(completed["status"], "succeeded");
    let log_response = send_json(
        &app.router,
        Method::GET,
        &format!("/api/v1/runs/{run_id}/logs"),
        None,
    )
    .await;
    let log = response_json(log_response).await;
    assert!(
        log["content"]
            .as_str()
            .expect("log content should be a string")
            .contains("shell-language-ok")
    );
}

#[tokio::test]
async fn manual_script_run_captures_exit_code_and_logs() {
    let app = test_app().await;
    let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
    let task = script_task(
        "echo_task",
        &working_dir,
        "cmd.exe",
        "/C",
        vec!["echo", "hello-from-task"],
        "0 0 2 * * *",
        5,
    );
    let create_response = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
    assert_eq!(create_response.status(), StatusCode::CREATED);

    let run_response = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks/echo_task/run",
        None,
    )
    .await;
    assert_eq!(run_response.status(), StatusCode::ACCEPTED);
    let run = response_json(run_response).await;
    let run_id = run["id"].as_i64().expect("run id should be an integer");
    assert_eq!(run["status"], "queued");

    let completed = wait_for_run(&app.router, run_id).await;
    assert_eq!(completed["status"], "succeeded");
    assert_eq!(completed["exit_code"], 0);
    assert_eq!(completed["trigger"], "manual");

    let log_response = send_json(
        &app.router,
        Method::GET,
        &format!("/api/v1/runs/{run_id}/logs"),
        None,
    )
    .await;
    assert_eq!(log_response.status(), StatusCode::OK);
    let log = response_json(log_response).await;
    assert_eq!(log["complete"], true);
    assert!(log["content"].as_str().unwrap().contains("hello-from-task"));

    let filtered_response = send_json(
        &app.router,
        Method::GET,
        "/api/v1/tasks/echo_task/runs?status=succeeded&trigger=manual",
        None,
    )
    .await;
    let filtered = response_json(filtered_response).await;
    assert_eq!(filtered["total"], 1);

    let download_response = send_json(
        &app.router,
        Method::GET,
        &format!("/api/v1/runs/{run_id}/logs/download"),
        None,
    )
    .await;
    assert_eq!(download_response.status(), StatusCode::OK);
    assert!(
        download_response
            .headers()
            .get(header::CONTENT_DISPOSITION)
            .and_then(|value| value.to_str().ok())
            .unwrap_or_default()
            .contains("attachment")
    );
    app.state.shutdown();
}

#[tokio::test]
async fn auto_detected_venv_injects_activation_environment() {
    let app = test_app().await;
    let project = app._temp_dir.path().join("project");
    let working_dir = project.join("src");
    let venv = project.join(".venv");
    std::fs::create_dir_all(&working_dir).unwrap();
    std::fs::create_dir_all(venv.join("Scripts")).unwrap();
    std::fs::write(venv.join("pyvenv.cfg"), "home = C:\\Python").unwrap();
    let command_shell = std::env::var_os("COMSPEC")
        .map(std::path::PathBuf::from)
        .unwrap_or_else(|| std::path::PathBuf::from("C:\\Windows\\System32\\cmd.exe"));
    std::fs::copy(command_shell, venv.join("Scripts").join("python.exe")).unwrap();

    let task = json!({
        "id": "venv_task",
        "name": "Venv task",
        "cron": "0 0 2 * * *",
        "timeout_secs": 5,
        "action": {
            "type": "script",
            "working_dir": working_dir,
            "interpreter": "python.exe",
            "script": "/C",
            "args": [
                "echo",
                "VIRTUAL=%VIRTUAL_ENV%;UNBUFFERED=%PYTHONUNBUFFERED%;HOME=%PYTHONHOME%;CUSTOM=%CUSTOM%"
            ],
            "env": {
                "PYTHONHOME": "C:\\wrong-python-home",
                "PYTHONUNBUFFERED": "0",
                "CUSTOM": "yes"
            }
        }
    });
    let create_response = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
    assert_eq!(create_response.status(), StatusCode::CREATED);
    let run = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks/venv_task/run",
        None,
    )
    .await;
    let run_id = response_json(run).await["id"].as_i64().unwrap();
    assert_eq!(
        wait_for_run(&app.router, run_id).await["status"],
        "succeeded"
    );
    let log = response_json(
        send_json(
            &app.router,
            Method::GET,
            &format!("/api/v1/runs/{run_id}/logs"),
            None,
        )
        .await,
    )
    .await;
    let content = log["content"].as_str().unwrap();
    assert!(
        content.contains(&format!(
            "VIRTUAL={};UNBUFFERED=1;HOME=%PYTHONHOME%;CUSTOM=yes",
            venv.display()
        )),
        "unexpected venv log output: {content}"
    );
    assert!(!content.contains("wrong-python-home"));
    app.state.shutdown();
}

#[tokio::test]
async fn http_and_process_executors_run_through_the_api() {
    let app = test_app().await;
    let (http_base, http_handle) = spawn_http_target().await;
    let create_http = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks",
        Some(http_task(&format!("{http_base}/check"))),
    )
    .await;
    assert_eq!(create_http.status(), StatusCode::CREATED);

    let http_run = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks/http_task/run",
        None,
    )
    .await;
    assert_eq!(http_run.status(), StatusCode::ACCEPTED);
    let http_run_id = response_json(http_run).await["id"].as_i64().unwrap();
    assert_eq!(
        wait_for_run(&app.router, http_run_id).await["status"],
        "succeeded"
    );

    let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
    let process_task = json!({
        "id": "process_task",
        "name": "Process task",
        "cron": "0 0 2 * * *",
        "timeout_secs": 5,
        "action": {
            "type": "process",
            "executable": "cmd.exe",
            "working_dir": working_dir,
            "args": ["/C", "echo", "process-output"],
            "env": {}
        }
    });
    let create_process = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks",
        Some(process_task),
    )
    .await;
    assert_eq!(create_process.status(), StatusCode::CREATED);
    let process_run = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks/process_task/run",
        None,
    )
    .await;
    let process_run_id = response_json(process_run).await["id"].as_i64().unwrap();
    assert_eq!(
        wait_for_run(&app.router, process_run_id).await["status"],
        "succeeded"
    );
    let process_log = response_json(
        send_json(
            &app.router,
            Method::GET,
            &format!("/api/v1/runs/{process_run_id}/logs"),
            None,
        )
        .await,
    )
    .await;
    assert!(
        process_log["content"]
            .as_str()
            .unwrap()
            .contains("process-output")
    );

    http_handle.abort();
    app.state.shutdown();
}

#[tokio::test]
async fn running_task_can_be_cancelled_and_sensitive_output_is_redacted() {
    let app = test_app().await;
    let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
    let task = json!({
        "id": "cancel_task",
        "name": "Cancel task",
        "cron": "0 0 2 * * *",
        "timeout_secs": 30,
        "action": {
            "type": "script",
            "working_dir": working_dir,
            "interpreter": "cmd.exe",
            "script": "/C",
            "args": ["ping", "-t", "127.0.0.1"],
            "env": {
                "API_TOKEN": "super-secret-token"
            }
        }
    });
    let create_response = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
    assert_eq!(create_response.status(), StatusCode::CREATED);

    let mut events = app.state.executor.subscribe();
    let run_response = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks/cancel_task/run",
        None,
    )
    .await;
    let run_id = response_json(run_response).await["id"].as_i64().unwrap();

    let cancel_response = send_json(
        &app.router,
        Method::POST,
        &format!("/api/v1/runs/{run_id}/cancel"),
        None,
    )
    .await;
    assert_eq!(cancel_response.status(), StatusCode::ACCEPTED);
    assert_eq!(
        wait_for_run(&app.router, run_id).await["status"],
        "cancelled"
    );

    let mut saw_run_event = false;
    let mut saw_log_event = false;
    for _ in 0..20 {
        let event = tokio::time::timeout(Duration::from_secs(1), events.recv())
            .await
            .expect("SSE event should arrive")
            .expect("SSE channel should stay open");
        match event {
            ExecutionEvent::Run { run } if run.id == run_id => saw_run_event = true,
            ExecutionEvent::Log {
                run_id: event_run_id,
                ..
            } if event_run_id == run_id => saw_log_event = true,
            _ => {}
        }
        if saw_run_event && saw_log_event {
            break;
        }
    }
    assert!(saw_run_event);
    assert!(saw_log_event);

    let terminal_cancel = send_json(
        &app.router,
        Method::POST,
        &format!("/api/v1/runs/{run_id}/cancel"),
        None,
    )
    .await;
    assert_eq!(terminal_cancel.status(), StatusCode::CONFLICT);

    let output_task = json!({
        "id": "redaction_task",
        "name": "Redaction task",
        "cron": "0 0 2 * * *",
        "timeout_secs": 5,
        "action": {
            "type": "script",
            "working_dir": app._temp_dir.path().to_string_lossy(),
            "interpreter": "cmd.exe",
            "script": "/C",
            "args": ["echo", "%API_TOKEN%"],
            "env": {
                "API_TOKEN": "super-secret-token"
            }
        }
    });
    let create_output = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks",
        Some(output_task),
    )
    .await;
    assert_eq!(create_output.status(), StatusCode::CREATED);
    let output_run = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks/redaction_task/run",
        None,
    )
    .await;
    let output_run_id = response_json(output_run).await["id"].as_i64().unwrap();
    wait_for_run(&app.router, output_run_id).await;
    let output_log = response_json(
        send_json(
            &app.router,
            Method::GET,
            &format!("/api/v1/runs/{output_run_id}/logs"),
            None,
        )
        .await,
    )
    .await;
    let content = output_log["content"].as_str().unwrap();
    assert!(content.contains("[REDACTED]"));
    assert!(!content.contains("super-secret-token"));
    app.state.shutdown();
}

#[tokio::test]
async fn overlapping_trigger_is_skipped_and_timeout_kills_process_tree() {
    let app = test_app().await;
    let marker_path = app._temp_dir.path().join("child-survived.txt");
    let marker = marker_path.to_string_lossy().replace('\'', "''");
    let script_path = app._temp_dir.path().join("timeout-test.ps1");
    let script = format!(
        "$marker = '{marker}'\n\
         $child = \"Start-Sleep -Seconds 3; Set-Content -LiteralPath '{marker}' -Value 'alive'\"\n\
         Start-Process -FilePath 'powershell.exe' -ArgumentList '-NoProfile', '-Command', $child -WindowStyle Hidden\n\
         Start-Sleep -Seconds 30\n"
    );
    std::fs::write(&script_path, script).expect("test script should be written");
    let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
    let task = script_task(
        "timeout_task",
        &working_dir,
        "powershell.exe",
        "-NoProfile",
        vec![
            "-ExecutionPolicy",
            "Bypass",
            "-File",
            &script_path.to_string_lossy(),
        ],
        "0 0 2 * * *",
        1,
    );
    let create_response = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
    assert_eq!(create_response.status(), StatusCode::CREATED);

    let first_response = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks/timeout_task/run",
        None,
    )
    .await;
    assert_eq!(first_response.status(), StatusCode::ACCEPTED);
    let first_run = response_json(first_response).await;
    let first_run_id = first_run["id"].as_i64().unwrap();

    let second_response = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks/timeout_task/run",
        None,
    )
    .await;
    assert_eq!(second_response.status(), StatusCode::ACCEPTED);
    let second_run = response_json(second_response).await;
    assert_eq!(second_run["status"], "skipped");

    let completed = wait_for_run(&app.router, first_run_id).await;
    assert_eq!(
        completed["status"], "timed_out",
        "unexpected run: {completed}"
    );
    assert!(
        completed["error_message"]
            .as_str()
            .unwrap()
            .contains("已终止进程树")
    );

    sleep(Duration::from_secs(3)).await;
    assert!(
        !marker_path.exists(),
        "child process survived after job termination"
    );
    app.state.shutdown();
}

#[tokio::test]
async fn scheduler_runs_tasks_and_tracks_enable_state() {
    let app = test_app().await;
    let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
    let task = script_task(
        "scheduled_task",
        &working_dir,
        "cmd.exe",
        "/C",
        vec!["exit", "0"],
        "*/1 * * * * *",
        5,
    );
    let create_response = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
    assert_eq!(create_response.status(), StatusCode::CREATED);
    assert!(app.state.scheduler.is_scheduled("scheduled_task"));

    let mut scheduled_run = None;
    for _ in 0..50 {
        let response = send_json(
            &app.router,
            Method::GET,
            "/api/v1/tasks/scheduled_task/runs",
            None,
        )
        .await;
        let runs = response_json(response).await;
        if let Some(item) = runs["items"]
            .as_array()
            .and_then(|items| items.first())
            .filter(|item| item["trigger"] == "schedule")
        {
            scheduled_run = Some(item.clone());
            break;
        }
        sleep(Duration::from_millis(100)).await;
    }
    let scheduled_run = scheduled_run.expect("scheduler should trigger a run");
    let run_id = scheduled_run["id"].as_i64().unwrap();
    assert_eq!(
        wait_for_run(&app.router, run_id).await["status"],
        "succeeded"
    );

    let disable_response = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks/scheduled_task/disable",
        None,
    )
    .await;
    assert_eq!(disable_response.status(), StatusCode::OK);
    assert!(!app.state.scheduler.is_scheduled("scheduled_task"));

    let enable_response = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks/scheduled_task/enable",
        None,
    )
    .await;
    assert_eq!(enable_response.status(), StatusCode::OK);
    assert!(app.state.scheduler.is_scheduled("scheduled_task"));
    app.state.shutdown();
}

#[tokio::test]
async fn scheduler_restores_enabled_tasks_after_restart() {
    let temp_dir = TempDir::new().expect("temporary directory should be created");
    let database_path = temp_dir.path().join("restart.db");
    let first_pool = connect_file(&database_path).await;
    let first_state = AppState::start(first_pool.clone(), temp_dir.path().join("logs-first"))
        .await
        .expect("first app state should start");
    let first_router = task_pilot_backend::router(first_state.clone());
    let working_dir = temp_dir.path().to_string_lossy().into_owned();
    let task = script_task(
        "restored_task",
        &working_dir,
        "cmd.exe",
        "/C",
        vec!["exit", "0"],
        "0 0 2 * * *",
        5,
    );
    let create_response = send_json(&first_router, Method::POST, "/api/v1/tasks", Some(task)).await;
    assert_eq!(create_response.status(), StatusCode::CREATED);
    first_state.shutdown();
    drop(first_router);
    drop(first_state);
    first_pool.close().await;

    let second_pool = connect_file(&database_path).await;
    let second_state = AppState::start(second_pool.clone(), temp_dir.path().join("logs-second"))
        .await
        .expect("second app state should start");
    assert!(second_state.scheduler.is_scheduled("restored_task"));
    second_state.shutdown();
    second_pool.close().await;
}

async fn run_command_task(app: &TestApp, id: &str, args: Vec<&str>) -> i64 {
    let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
    let task = script_task(id, &working_dir, "cmd.exe", "/C", args, "0 0 2 * * *", 10);
    let create_response = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
    assert_eq!(create_response.status(), StatusCode::CREATED);
    let run_response = send_json(
        &app.router,
        Method::POST,
        &format!("/api/v1/tasks/{id}/run"),
        None,
    )
    .await;
    assert_eq!(run_response.status(), StatusCode::ACCEPTED);
    response_json(run_response).await["id"]
        .as_i64()
        .expect("run id should be an integer")
}

async fn read_log_chunk(app: &TestApp, run_id: i64, offset: u64) -> Value {
    let response = send_json(
        &app.router,
        Method::GET,
        &format!("/api/v1/runs/{run_id}/logs?offset={offset}"),
        None,
    )
    .await;
    assert_eq!(response.status(), StatusCode::OK);

    response_json(response).await
}

#[tokio::test]
async fn gbk_encoded_output_is_decoded_to_utf8() {
    let app = test_app().await;
    // 固定的 GBK 字节只在中日韩等使用该代码页的系统上才有意义。
    let encoding = task_pilot_backend::runs::executor::OutputEncoding::from_env();
    if encoding.name() != "GBK" {
        return;
    }

    let payload = app._temp_dir.path().join("gbk.txt");
    // "中文" 的 GBK 编码
    std::fs::write(&payload, [0xD6, 0xD0, 0xCE, 0xC4]).expect("GBK fixture should be written");
    let run_id = run_command_task(&app, "gbk_task", vec!["type", "gbk.txt"]).await;

    let completed = wait_for_run(&app.router, run_id).await;
    assert_eq!(completed["status"], "succeeded");
    let log = read_log_chunk(&app, run_id, 0).await;
    let content = log["content"].as_str().expect("log content should be text");

    assert!(
        content.contains("中文"),
        "unexpected log content: {content}"
    );
    assert!(!content.contains('\u{fffd}'));
    app.state.shutdown();
}

#[tokio::test]
async fn long_utf8_log_is_reassembled_across_chunk_boundaries() {
    let app = test_app().await;
    let payload = app._temp_dir.path().join("long.txt");
    let body = format!("{}\n尾巴\n", "中".repeat(30_000));
    std::fs::write(&payload, body.as_bytes()).expect("UTF-8 fixture should be written");
    let run_id = run_command_task(&app, "long_log_task", vec!["type", "long.txt"]).await;

    let completed = wait_for_run(&app.router, run_id).await;
    assert_eq!(completed["status"], "succeeded");

    let first = read_log_chunk(&app, run_id, 0).await;
    assert_eq!(
        first["complete"], false,
        "a terminal run must not report completion before the whole log is read"
    );
    let mut content = first["content"].as_str().unwrap_or_default().to_owned();
    let mut offset = first["next_offset"]
        .as_u64()
        .expect("next_offset should be an integer");

    for _ in 0..20 {
        let chunk = read_log_chunk(&app, run_id, offset).await;
        content.push_str(chunk["content"].as_str().unwrap_or_default());
        offset = chunk["next_offset"]
            .as_u64()
            .expect("next_offset should be an integer");
        if chunk["complete"].as_bool().unwrap_or(false) {
            break;
        }
    }

    assert!(
        !content.contains('\u{fffd}'),
        "log lost a character boundary"
    );
    assert!(content.contains("尾巴"));
    assert_eq!(
        content.len(),
        usize::try_from(offset).expect("offset fits usize")
    );
    app.state.shutdown();
}

#[tokio::test]
async fn stats_endpoint_aggregates_task_counts() {
    let app = test_app().await;
    let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
    let script = script_task(
        "stats_script",
        &working_dir,
        "cmd.exe",
        "/C",
        vec!["exit", "0"],
        "0 0 2 * * *",
        5,
    );
    let http = http_task("http://127.0.0.1:9/never");
    for payload in [script.clone(), http.clone()] {
        let response = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(payload)).await;
        assert_eq!(response.status(), StatusCode::CREATED);
    }

    let response = send_json(&app.router, Method::GET, "/api/v1/stats", None).await;
    assert_eq!(response.status(), StatusCode::OK);
    let stats = response_json(response).await;

    assert_eq!(stats["total"], 2);
    assert_eq!(stats["enabled"], 2);
    assert_eq!(stats["script"], 1);
    assert_eq!(stats["http"], 1);
    assert_eq!(stats["process"], 0);
    assert_eq!(stats["running"], 0);
    app.state.shutdown();
}

#[tokio::test]
async fn schedule_endpoint_reports_next_run_for_enabled_tasks() {
    let app = test_app().await;
    let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
    let task = script_task(
        "scheduled_stats",
        &working_dir,
        "cmd.exe",
        "/C",
        vec!["exit", "0"],
        "*/30 * * * * *",
        5,
    );
    let created = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
    assert_eq!(created.status(), StatusCode::CREATED);

    let response = send_json(&app.router, Method::GET, "/api/v1/schedule", None).await;
    assert_eq!(response.status(), StatusCode::OK);
    let schedule = response_json(response).await;
    let entries = schedule.as_array().expect("schedule should be an array");
    let entry = entries
        .iter()
        .find(|item| item["task_id"] == "scheduled_stats")
        .expect("enabled task should appear in the schedule");

    assert!(
        entry["next_run_at"].is_string(),
        "next_run_at should be computed for a registered task"
    );
    app.state.shutdown();
}

#[tokio::test]
async fn http_tasks_are_registered_for_scheduling() {
    let app = test_app().await;
    let task = http_task("http://127.0.0.1:9/never");
    let created = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
    assert_eq!(created.status(), StatusCode::CREATED);

    assert!(
        app.state.scheduler.is_scheduled("http_task"),
        "HTTP tasks must take part in scheduled runs"
    );

    let disabled = send_json(
        &app.router,
        Method::POST,
        "/api/v1/tasks/http_task/disable",
        None,
    )
    .await;
    assert_eq!(disabled.status(), StatusCode::OK);
    assert!(!app.state.scheduler.is_scheduled("http_task"));
    app.state.shutdown();
}

#[tokio::test]
async fn startup_marks_runs_interrupted_by_restart_as_cancelled() {
    let app = test_app().await;
    let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
    let task = script_task(
        "interrupted_task",
        &working_dir,
        "cmd.exe",
        "/C",
        vec!["exit", "0"],
        "0 0 2 * * *",
        5,
    );
    let created = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
    assert_eq!(created.status(), StatusCode::CREATED);

    // 模拟上一次进程被强杀后遗留的运行记录。
    sqlx::query(
        "INSERT INTO task_runs (task_id, trigger_type, status, created_at)
         VALUES ('interrupted_task', 'manual', 'running', ?)",
    )
    .bind(chrono::Utc::now())
    .execute(&app.state.pool)
    .await
    .expect("stale run should be inserted");

    let reconciled = task_pilot_backend::runs::repository::fail_interrupted_runs(&app.state.pool)
        .await
        .expect("reconciliation should run");
    assert_eq!(reconciled, 1);

    let runs = send_json(
        &app.router,
        Method::GET,
        "/api/v1/tasks/interrupted_task/runs",
        None,
    )
    .await;
    let body = response_json(runs).await;
    let item = &body["items"][0];
    assert_eq!(item["status"], "cancelled");
    assert!(
        item["error_message"]
            .as_str()
            .unwrap_or_default()
            .contains("重启")
    );
    app.state.shutdown();
}