task_pilot/backend/src/runs/executor.rs

use std::{
    collections::{BTreeMap, HashMap},
    path::{Path, PathBuf},
    process::Stdio,
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, AtomicU64, Ordering},
    },
    time::{Instant, SystemTime},
};

use chrono::Utc;
use encoding_rs::Encoding;
use futures_util::StreamExt;
use reqwest::Method;
use serde::Serialize;
use sqlx::SqlitePool;
use tokio::{
    io::{AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncWriteExt, BufReader},
    process::{Child, Command},
    sync::{Mutex as AsyncMutex, broadcast},
    time::{Duration, sleep, timeout},
};
use tokio_util::sync::CancellationToken;

use crate::{
    error::ApiResult,
    process_job::ProcessJob,
    python_env::{is_python_interpreter, prepare_script_environment},
    runs::{
        model::{RunStatus, RunTrigger, TaskRunResponse},
        repository,
    },
    tasks::model::{HttpMethod, TaskAction, TaskResponse},
};

const MAX_LOG_BYTES: u64 = 5 * 1024 * 1024;
const MAX_LOG_LINE_BYTES: usize = 256 * 1024;
const MAX_HTTP_LOG_BODY_BYTES: usize = 64 * 1024;
/// 合并推送的日志事件上限：攒够这么多字节就立即发一个事件。
const LOG_EVENT_BATCH_BYTES: usize = 16 * 1024;
/// 合并推送的时间窗口：超过这个时间还没攒满也发出去，保证低输出时也有实时感。
const LOG_EVENT_BATCH_INTERVAL: Duration = Duration::from_millis(100);
const TERMINATION_WAIT_SECS: u64 = 5;
const CREATE_NO_WINDOW: u32 = 0x0800_0000;
const DEFAULT_RETENTION_DAYS: u64 = 30;
const DEFAULT_MAX_TOTAL_LOG_BYTES: u64 = 512 * 1024 * 1024;
const LOG_ENCODING_ENV: &str = "TASK_PILOT_LOG_ENCODING";

/// 子进程输出的解码方式。
///
/// `TASK_PILOT_LOG_ENCODING=auto`（默认）逐行先按 UTF-8 严格解码，失败时回退到系统
/// ANSI 代码页；填写编码标签（`gbk`、`utf-8`、`windows-1251` 等）则始终按该编码解码。
#[derive(Clone, Copy, Debug)]
pub struct OutputEncoding {
    forced: Option<&'static Encoding>,
    fallback: &'static Encoding,
}

impl OutputEncoding {
    pub fn from_env() -> Self {
        let fallback = ansi_code_page_encoding();
        let forced = std::env::var(LOG_ENCODING_ENV).ok().and_then(|value| {
            let label = value.trim().to_ascii_lowercase();
            if label.is_empty() || label == "auto" {
                None
            } else {
                Encoding::for_label(label.as_bytes())
            }
        });

        Self { forced, fallback }
    }

    pub fn decode(&self, bytes: &[u8]) -> String {
        if let Some(encoding) = self.forced {
            let (text, _) = encoding.decode_without_bom_handling(bytes);
            return text.into_owned();
        }

        match std::str::from_utf8(bytes) {
            Ok(text) => text.to_owned(),
            Err(_) => {
                let (text, _) = self.fallback.decode_without_bom_handling(bytes);
                text.into_owned()
            }
        }
    }

    /// 实际使用的编码名，写入运行日志便于排查乱码。
    pub fn name(&self) -> &'static str {
        self.forced.unwrap_or(self.fallback).name()
    }
}

#[cfg(windows)]
fn ansi_code_page_encoding() -> &'static Encoding {
    let code_page = unsafe { windows_sys::Win32::Globalization::GetACP() };
    code_page_encoding(code_page)
}

#[cfg(not(windows))]
fn ansi_code_page_encoding() -> &'static Encoding {
    encoding_rs::UTF_8
}

fn code_page_encoding(code_page: u32) -> &'static Encoding {
    let label = match code_page {
        936 => "gbk".to_owned(),
        950 => "big5".to_owned(),
        932 => "shift_jis".to_owned(),
        949 => "euc-kr".to_owned(),
        65001 => "utf-8".to_owned(),
        other => format!("windows-{other}"),
    };

    Encoding::for_label(label.as_bytes()).unwrap_or(encoding_rs::UTF_8)
}

#[derive(Clone, Debug)]
pub struct LogRetentionConfig {
    pub retention_days: u64,
    pub max_total_bytes: u64,
}

impl Default for LogRetentionConfig {
    fn default() -> Self {
        Self {
            retention_days: DEFAULT_RETENTION_DAYS,
            max_total_bytes: DEFAULT_MAX_TOTAL_LOG_BYTES,
        }
    }
}

#[derive(Clone, Debug, Serialize)]
#[serde(tag = "event", content = "data", rename_all = "snake_case")]
pub enum ExecutionEvent {
    Run {
        run: TaskRunResponse,
    },
    Log {
        run_id: i64,
        offset: u64,
        next_offset: u64,
        content: String,
    },
}

impl ExecutionEvent {
    pub const fn event_name(&self) -> &'static str {
        match self {
            Self::Run { .. } => "run",
            Self::Log { .. } => "log",
        }
    }
}

#[derive(Clone)]
pub struct TaskExecutor {
    inner: Arc<ExecutorInner>,
}

struct ExecutorInner {
    pool: SqlitePool,
    logs_dir: PathBuf,
    shutdown: CancellationToken,
    running: Mutex<HashMap<String, RunningHandle>>,
    events: broadcast::Sender<ExecutionEvent>,
    http: reqwest::Client,
    log_config: LogRetentionConfig,
    cleanup_lock: AsyncMutex<()>,
    output_encoding: OutputEncoding,
}

struct RunningHandle {
    run_id: Option<i64>,
    cancellation: CancellationToken,
}

#[derive(Clone)]
struct LogSink {
    inner: Arc<LogSinkInner>,
}

struct LogSinkInner {
    file: AsyncMutex<tokio::fs::File>,
    size: AtomicU64,
    truncated: AtomicBool,
    run_id: i64,
    events: broadcast::Sender<ExecutionEvent>,
    secrets: Vec<String>,
    encoding: OutputEncoding,
    /// 已经写进文件、但还没通过 SSE 推送出去的字节。
    pending: Mutex<PendingBatch>,
}

/// 等待合并推送的一段日志。`offset` 是这段内容在日志文件中的起始偏移。
#[derive(Default)]
struct PendingBatch {
    offset: u64,
    bytes: usize,
    opened_at: Option<Instant>,
    text: String,
}

struct ScriptOutcome {
    status: RunStatus,
    exit_code: Option<i32>,
    error_message: Option<String>,
}

enum ProcessEnd {
    Exited(std::process::ExitStatus),
    TimedOut,
    Cancelled,
}

impl TaskExecutor {
    pub fn new(pool: SqlitePool, logs_dir: PathBuf, log_config: LogRetentionConfig) -> Self {
        let (events, _) = broadcast::channel(512);
        let output_encoding = OutputEncoding::from_env();
        tracing::info!(encoding = output_encoding.name(), "task output decoding");
        Self {
            inner: Arc::new(ExecutorInner {
                pool,
                logs_dir,
                shutdown: CancellationToken::new(),
                running: Mutex::new(HashMap::new()),
                events,
                http: reqwest::Client::builder()
                    .user_agent("TaskPilot/0.1")
                    .redirect(reqwest::redirect::Policy::limited(10))
                    .build()
                    .unwrap_or_default(),
                log_config,
                cleanup_lock: AsyncMutex::new(()),
                output_encoding,
            }),
        }
    }

    pub async fn trigger(
        &self,
        task: &TaskResponse,
        trigger: RunTrigger,
    ) -> ApiResult<TaskRunResponse> {
        let cancellation = CancellationToken::new();
        let already_running = {
            let mut running = lock_running(&self.inner.running);
            if running.contains_key(&task.id) {
                true
            } else {
                running.insert(
                    task.id.clone(),
                    RunningHandle {
                        run_id: None,
                        cancellation: cancellation.clone(),
                    },
                );
                false
            }
        };
        if already_running {
            let run = repository::create_skipped(
                &self.inner.pool,
                &task.id,
                trigger,
                "同一任务已有运行实例，本次触发已跳过",
            )
            .await?;
            self.emit_run(&run);
            return Ok(run);
        }

        let run = match repository::create_queued(&self.inner.pool, &task.id, trigger).await {
            Ok(run) => run,
            Err(error) => {
                remove_running_if(&self.inner.running, &task.id, None);
                return Err(error);
            }
        };
        if let Some(handle) = lock_running(&self.inner.running).get_mut(&task.id) {
            handle.run_id = Some(run.id);
        }
        self.emit_run(&run);

        let executor = self.clone();
        let task = task.clone();
        tokio::spawn(async move {
            executor.execute_run(task, run.id, cancellation).await;
        });

        Ok(run)
    }

    pub fn cancel_task(&self, task_id: &str) -> bool {
        let mut running = lock_running(&self.inner.running);
        if let Some(handle) = running.remove(task_id) {
            handle.cancellation.cancel();
            true
        } else {
            false
        }
    }

    pub fn cancel_run(&self, run_id: i64) -> bool {
        let mut running = lock_running(&self.inner.running);
        let task_id = running
            .iter()
            .find_map(|(task_id, handle)| (handle.run_id == Some(run_id)).then(|| task_id.clone()));
        if let Some(task_id) = task_id
            && let Some(handle) = running.remove(&task_id)
        {
            handle.cancellation.cancel();
            return true;
        }
        false
    }

    pub fn subscribe(&self) -> broadcast::Receiver<ExecutionEvent> {
        self.inner.events.subscribe()
    }

    /// 当前正在执行的任务数量（用于统计接口）。
    pub fn running_count(&self) -> usize {
        lock_running(&self.inner.running).len()
    }

    pub fn log_path(&self, task_id: &str, run_id: i64) -> PathBuf {
        self.inner
            .logs_dir
            .join(task_id)
            .join(format!("{run_id}.log"))
    }

    pub fn shutdown(&self) {
        self.inner.shutdown.cancel();
        let running = lock_running(&self.inner.running);
        for handle in running.values() {
            handle.cancellation.cancel();
        }
    }

    pub async fn cleanup_logs(&self) -> std::io::Result<()> {
        let _guard = self.inner.cleanup_lock.lock().await;
        let logs_dir = self.inner.logs_dir.clone();
        let config = self.inner.log_config.clone();
        let active = self.active_log_paths();
        tokio::task::spawn_blocking(move || cleanup_logs_sync(&logs_dir, &config, &active))
            .await
            .map_err(std::io::Error::other)?
    }

    fn active_log_paths(&self) -> std::collections::HashSet<PathBuf> {
        lock_running(&self.inner.running)
            .iter()
            .filter_map(|(task_id, handle)| {
                handle.run_id.map(|run_id| {
                    self.inner
                        .logs_dir
                        .join(task_id)
                        .join(format!("{run_id}.log"))
                })
            })
            .collect()
    }

    async fn execute_run(&self, task: TaskResponse, run_id: i64, cancellation: CancellationToken) {
        let started_at = Utc::now();
        let started = Instant::now();
        match repository::mark_running(&self.inner.pool, run_id, started_at).await {
            Ok(true) => {}
            Ok(false) => {
                tracing::warn!(
                    run_id,
                    "run row disappeared before it could be marked running"
                );
                remove_running_if(&self.inner.running, &task.id, Some(run_id));
                return;
            }
            Err(error) => {
                tracing::error!(run_id, error = %error, "failed to mark task run as running");
                remove_running_if(&self.inner.running, &task.id, Some(run_id));
                return;
            }
        }
        self.emit_run_by_id(run_id).await;

        let outcome = self.run_action(&task, run_id, &cancellation).await;
        let finished_at = Utc::now();
        let duration_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);

        match repository::complete(
            &self.inner.pool,
            run_id,
            outcome.status,
            finished_at,
            duration_ms,
            outcome.exit_code,
            outcome.error_message.as_deref(),
        )
        .await
        {
            Ok(true) => {}
            Ok(false) => {
                tracing::warn!(
                    run_id,
                    "run row disappeared before completion could be recorded"
                );
            }
            Err(error) => tracing::error!(run_id, error = %error, "failed to complete task run"),
        }

        self.emit_run_by_id(run_id).await;
        // 只清理属于自己的句柄：取消后立刻重跑时，旧 run 收尾不能删掉新 run 的闸门。
        remove_running_if(&self.inner.running, &task.id, Some(run_id));
        if let Err(error) = self.cleanup_logs().await {
            tracing::warn!(%error, "failed to clean task logs");
        }
    }

    async fn run_action(
        &self,
        task: &TaskResponse,
        run_id: i64,
        cancellation: &CancellationToken,
    ) -> ScriptOutcome {
        match &task.action {
            TaskAction::Script {
                working_dir,
                venv,
                interpreter,
                script,
                args,
                env,
            } => {
                let prepared = match prepare_script_environment(
                    working_dir,
                    interpreter,
                    venv.as_deref(),
                    env,
                ) {
                    Ok(prepared) => prepared,
                    Err(error) => {
                        return failed_outcome(format!("准备 Python 运行环境失败: {error}"));
                    }
                };
                let mut command = Command::new(&prepared.interpreter);
                prepared.apply(&mut command);

                let mut notes = Vec::new();
                if let Some(venv) = &prepared.virtual_env {
                    notes.push(format!("使用 Python 虚拟环境: {}", venv.display()));
                }
                if is_python_interpreter(interpreter) {
                    match crate::python_mirrors::repository::default_enabled(&self.inner.pool).await
                    {
                        Ok(Some(mirror)) => {
                            if !env_has(env, "PIP_INDEX_URL") {
                                command.env("PIP_INDEX_URL", &mirror.index_url);
                            }
                            if !env_has(env, "UV_INDEX_URL") {
                                command.env("UV_INDEX_URL", &mirror.index_url);
                            }
                            if let Some(host) = mirror
                                .trusted_host
                                .as_deref()
                                .map(str::trim)
                                .filter(|value| !value.is_empty())
                                && !env_has(env, "PIP_TRUSTED_HOST")
                            {
                                command.env("PIP_TRUSTED_HOST", host);
                            }
                            notes.push(format!(
                                "Python 包镜像: {} ({})",
                                mirror.name, mirror.index_url
                            ));
                        }
                        Ok(None) => {}
                        Err(error) => {
                            tracing::warn!(
                                task_id = %task.id,
                                %error,
                                "failed to load default python mirror"
                            );
                        }
                    }
                }

                command.arg(script).args(args).current_dir(working_dir);
                let startup_note = (!notes.is_empty()).then(|| notes.join("；"));
                self.run_command(task, run_id, cancellation, command, startup_note)
                    .await
            }
            TaskAction::Process {
                executable,
                working_dir,
                args,
                env,
            } => {
                let mut command = Command::new(executable);
                command.args(args).envs(env);
                if let Some(working_dir) = working_dir {
                    command.current_dir(working_dir);
                }
                self.run_command(task, run_id, cancellation, command, None)
                    .await
            }
            TaskAction::Http { .. } => self.run_http(task, run_id, cancellation).await,
        }
    }

    async fn run_command(
        &self,
        task: &TaskResponse,
        run_id: i64,
        cancellation: &CancellationToken,
        mut command: Command,
        startup_note: Option<String>,
    ) -> ScriptOutcome {
        let log_path = self.log_path(&task.id, run_id);
        let secrets = sensitive_values(&task.action);
        let sink = match LogSink::create(
            &log_path,
            &task.id,
            run_id,
            self.inner.events.clone(),
            secrets,
            self.inner.output_encoding,
        )
        .await
        {
            Ok(sink) => sink,
            Err(error) => return failed_outcome(format!("无法创建运行日志: {error}")),
        };
        if let Some(note) = startup_note {
            sink.system(&note).await;
        }

        let job = match ProcessJob::new() {
            Ok(job) => job,
            Err(error) => {
                sink.system(&format!("无法创建 Windows Job Object: {error}"))
                    .await;
                return failed_outcome(format!("无法创建 Windows Job Object: {error}"));
            }
        };

        command
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);
        #[cfg(windows)]
        {
            command.creation_flags(CREATE_NO_WINDOW);
        }

        let mut child = match command.spawn() {
            Ok(child) => child,
            Err(error) => {
                sink.system(&format!("启动进程失败: {error}")).await;
                return failed_outcome(format!("启动进程失败: {error}"));
            }
        };

        if let Err(error) = job.assign(&child) {
            let _ = child.kill().await;
            sink.system(&format!("关联 Job Object 失败: {error}")).await;
            return failed_outcome(format!("关联 Windows Job Object 失败: {error}"));
        }

        let stdout_task = child
            .stdout
            .take()
            .map(|stdout| spawn_log_reader(stdout, "stdout", sink.clone()));
        let stderr_task = child
            .stderr
            .take()
            .map(|stderr| spawn_log_reader(stderr, "stderr", sink.clone()));

        let process_end = tokio::select! {
            status = child.wait() => {
                match status {
                    Ok(status) => ProcessEnd::Exited(status),
                    Err(error) => {
                        sink.system(&format!("等待进程失败: {error}")).await;
                        return failed_outcome(format!("等待进程失败: {error}"));
                    }
                }
            }
            _ = sleep(Duration::from_secs(task.timeout_secs)) => {
                if let Err(error) = job.terminate() {
                    tracing::error!(task_id = %task.id, run_id, %error, "failed to terminate timed out job");
                }
                wait_after_termination(&mut child).await;
                ProcessEnd::TimedOut
            }
            _ = cancellation.cancelled() => {
                if let Err(error) = job.terminate() {
                    tracing::error!(task_id = %task.id, run_id, %error, "failed to terminate cancelled job");
                }
                wait_after_termination(&mut child).await;
                ProcessEnd::Cancelled
            }
            _ = self.inner.shutdown.cancelled() => {
                if let Err(error) = job.terminate() {
                    tracing::error!(task_id = %task.id, run_id, %error, "failed to terminate job during shutdown");
                }
                wait_after_termination(&mut child).await;
                ProcessEnd::Cancelled
            }
        };

        if let Some(stdout_task) = stdout_task
            && let Err(error) = stdout_task.await
        {
            tracing::warn!(task_id = %task.id, run_id, %error, "stdout capture task failed");
        }
        if let Some(stderr_task) = stderr_task
            && let Err(error) = stderr_task.await
        {
            tracing::warn!(task_id = %task.id, run_id, %error, "stderr capture task failed");
        }

        match process_end {
            ProcessEnd::Exited(status) => {
                let exit_code = status.code();
                if status.success() {
                    sink.system("任务执行成功").await;
                    ScriptOutcome {
                        status: RunStatus::Succeeded,
                        exit_code,
                        error_message: None,
                    }
                } else {
                    let message = format!(
                        "进程退出码为 {}",
                        exit_code.map_or_else(|| "unknown".to_owned(), |code| code.to_string())
                    );
                    sink.system(&message).await;
                    ScriptOutcome {
                        status: RunStatus::Failed,
                        exit_code,
                        error_message: Some(message),
                    }
                }
            }
            ProcessEnd::TimedOut => {
                let message = format!("任务执行超过 {} 秒，已终止进程树", task.timeout_secs);
                sink.system(&message).await;
                ScriptOutcome {
                    status: RunStatus::TimedOut,
                    exit_code: None,
                    error_message: Some(message),
                }
            }
            ProcessEnd::Cancelled => {
                let message = "任务已取消".to_owned();
                sink.system(&message).await;
                ScriptOutcome {
                    status: RunStatus::Cancelled,
                    exit_code: None,
                    error_message: Some(message),
                }
            }
        }
    }

    async fn run_http(
        &self,
        task: &TaskResponse,
        run_id: i64,
        cancellation: &CancellationToken,
    ) -> ScriptOutcome {
        let TaskAction::Http {
            url,
            method,
            headers,
            body,
            expected_status,
        } = &task.action
        else {
            return failed_outcome("invalid HTTP action payload");
        };

        let log_path = self.log_path(&task.id, run_id);
        let sink = match LogSink::create(
            &log_path,
            &task.id,
            run_id,
            self.inner.events.clone(),
            sensitive_values(&task.action),
            self.inner.output_encoding,
        )
        .await
        {
            Ok(sink) => sink,
            Err(error) => return failed_outcome(format!("无法创建运行日志: {error}")),
        };

        let method = match method {
            HttpMethod::Get => Method::GET,
            HttpMethod::Post => Method::POST,
        };
        let mut request = self
            .inner
            .http
            .request(method.clone(), url)
            .timeout(Duration::from_secs(task.timeout_secs));
        for (name, value) in headers {
            request = request.header(name, value);
        }
        if let Some(body) = body {
            request = request.body(body.clone());
        }

        sink.http(&format!("请求 {method} {url}")).await;
        let response = tokio::select! {
            response = request.send() => {
                match response {
                    Ok(response) => response,
                    Err(error) => {
                        let message = if error.is_timeout() {
                            format!("HTTP 请求超过 {} 秒", task.timeout_secs)
                        } else {
                            format!("HTTP 请求失败: {error}")
                        };
                        sink.system(&message).await;
                        return ScriptOutcome {
                            status: if error.is_timeout() {
                                RunStatus::TimedOut
                            } else {
                                RunStatus::Failed
                            },
                            exit_code: None,
                            error_message: Some(message),
                        };
                    }
                }
            }
            _ = cancellation.cancelled() => {
                sink.system("HTTP 请求已取消").await;
                return ScriptOutcome {
                    status: RunStatus::Cancelled,
                    exit_code: None,
                    error_message: Some("HTTP 请求已取消".to_owned()),
                };
            }
            _ = self.inner.shutdown.cancelled() => {
                sink.system("HTTP 请求因服务关闭而取消").await;
                return ScriptOutcome {
                    status: RunStatus::Cancelled,
                    exit_code: None,
                    error_message: Some("HTTP 请求因服务关闭而取消".to_owned()),
                };
            }
        };

        let http_status = response.status();
        sink.http(&format!("响应状态 {http_status}")).await;
        let mut response_stream = response.bytes_stream();
        let mut response_body = Vec::new();
        let mut body_truncated = false;
        loop {
            let next = tokio::select! {
                chunk = response_stream.next() => chunk,
                _ = cancellation.cancelled() => {
                    sink.system("读取 HTTP 响应已取消").await;
                    return ScriptOutcome {
                        status: RunStatus::Cancelled,
                        exit_code: None,
                        error_message: Some("读取 HTTP 响应已取消".to_owned()),
                    };
                }
                _ = self.inner.shutdown.cancelled() => {
                    sink.system("HTTP 响应读取因服务关闭而取消").await;
                    return ScriptOutcome {
                        status: RunStatus::Cancelled,
                        exit_code: None,
                        error_message: Some("HTTP 响应读取因服务关闭而取消".to_owned()),
                    };
                }
            };

            match next {
                Some(Ok(chunk)) => {
                    let remaining = MAX_HTTP_LOG_BODY_BYTES.saturating_sub(response_body.len());
                    let copy_length = remaining.min(chunk.len());
                    response_body.extend_from_slice(&chunk[..copy_length]);
                    if chunk.len() > copy_length {
                        body_truncated = true;
                        break;
                    }
                }
                Some(Err(error)) => {
                    let message = if error.is_timeout() {
                        format!("读取 HTTP 响应超过 {} 秒", task.timeout_secs)
                    } else {
                        format!("读取 HTTP 响应失败: {error}")
                    };
                    sink.system(&message).await;
                    return ScriptOutcome {
                        status: if error.is_timeout() {
                            RunStatus::TimedOut
                        } else {
                            RunStatus::Failed
                        },
                        exit_code: None,
                        error_message: Some(message),
                    };
                }
                None => break,
            }
        }
        let body_text = self.inner.output_encoding.decode(&response_body);
        if !body_text.is_empty() {
            sink.http(&body_text).await;
        }
        if body_truncated {
            sink.system(&format!(
                "HTTP 响应体超过 {MAX_HTTP_LOG_BODY_BYTES} 字节，日志仅保留前部内容"
            ))
            .await;
        }

        let expected = if expected_status.is_empty() {
            http_status.is_success()
        } else {
            expected_status.contains(&http_status.as_u16())
        };
        if expected {
            sink.system("HTTP 任务执行成功").await;
            ScriptOutcome {
                status: RunStatus::Succeeded,
                exit_code: Some(i32::from(http_status.as_u16())),
                error_message: None,
            }
        } else {
            let message = format!("HTTP 状态码 {http_status} 不在期望范围内");
            sink.system(&message).await;
            ScriptOutcome {
                status: RunStatus::Failed,
                exit_code: Some(i32::from(http_status.as_u16())),
                error_message: Some(message),
            }
        }
    }

    fn emit_run(&self, run: &TaskRunResponse) {
        let _ = self
            .inner
            .events
            .send(ExecutionEvent::Run { run: run.clone() });
    }

    async fn emit_run_by_id(&self, run_id: i64) {
        match repository::get(&self.inner.pool, run_id).await {
            Ok(Some(run)) => self.emit_run(&run),
            Ok(None) => {}
            Err(error) => {
                tracing::warn!(run_id, error = %error, "failed to load run for SSE event");
            }
        }
    }
}

impl LogSink {
    async fn create(
        path: &Path,
        task_id: &str,
        run_id: i64,
        events: broadcast::Sender<ExecutionEvent>,
        secrets: Vec<String>,
        encoding: OutputEncoding,
    ) -> std::io::Result<Self> {
        if let Some(parent) = path.parent() {
            tokio::fs::create_dir_all(parent).await?;
        }
        let header = format!(
            "[system] TaskPilot run {run_id} for task {task_id} started at {}\n",
            Utc::now().to_rfc3339()
        );
        tokio::fs::write(path, header.as_bytes()).await?;
        let file = tokio::fs::OpenOptions::new()
            .append(true)
            .open(path)
            .await?;
        let mut secrets = secrets;
        secrets.sort_by_key(|value| std::cmp::Reverse(value.len()));

        let inner = Arc::new(LogSinkInner {
            file: AsyncMutex::new(file),
            size: AtomicU64::new(header.len() as u64),
            truncated: AtomicBool::new(false),
            run_id,
            events,
            secrets,
            encoding,
            pending: Mutex::new(PendingBatch::default()),
        });
        spawn_pending_flusher(&inner);

        Ok(Self { inner })
    }

    fn decode_output(&self, bytes: &[u8]) -> String {
        self.inner.encoding.decode(bytes)
    }

    async fn append(&self, text: &str) {
        let redacted = redact_text(text, &self.inner.secrets);
        let bytes = redacted.as_bytes();
        let mut file = self.inner.file.lock().await;
        let current = self.inner.size.load(Ordering::Relaxed);
        if current + bytes.len() as u64 > MAX_LOG_BYTES {
            if !self.inner.truncated.swap(true, Ordering::Relaxed) {
                let notice = format!("[system] 日志超过 {MAX_LOG_BYTES} 字节，后续输出已丢弃\n");
                let _ = file.write_all(notice.as_bytes()).await;
            }
            return;
        }
        if file.write_all(bytes).await.is_err() {
            return;
        }
        let next = current + bytes.len() as u64;
        self.inner.size.store(next, Ordering::Relaxed);

        // 文件写入是串行的，所以 pending 也在文件锁内更新，保证推送的偏移连续。
        let mut pending = lock_pending(&self.inner.pending);
        if pending.bytes == 0 {
            pending.offset = current;
            pending.opened_at = Some(Instant::now());
        }
        pending.bytes += bytes.len();
        pending.text.push_str(&redacted);

        let batch_full = pending.bytes >= LOG_EVENT_BATCH_BYTES;
        let window_elapsed = pending
            .opened_at
            .is_some_and(|opened| opened.elapsed() >= LOG_EVENT_BATCH_INTERVAL);
        if batch_full || window_elapsed {
            emit_pending(&self.inner, &mut pending);
        }
    }

    async fn system(&self, message: &str) {
        self.append(&format!("[system] {message}\n")).await;
    }

    async fn http(&self, message: &str) {
        self.append(&format!("[http] {message}\n")).await;
    }
}

impl Drop for LogSinkInner {
    fn drop(&mut self) {
        let mut pending = lock_pending(&self.pending);
        emit_pending(self, &mut pending);
    }
}

fn lock_pending(pending: &Mutex<PendingBatch>) -> std::sync::MutexGuard<'_, PendingBatch> {
    pending
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
}

/// 把待推送的日志合并成一个事件发出去，并清空缓冲区。
fn emit_pending(inner: &LogSinkInner, pending: &mut PendingBatch) {
    if pending.bytes == 0 {
        return;
    }
    let batch = std::mem::take(pending);
    let _ = inner.events.send(ExecutionEvent::Log {
        run_id: inner.run_id,
        offset: batch.offset,
        next_offset: batch.offset + batch.bytes as u64,
        content: batch.text,
    });
}

fn flush_pending_if_due(inner: &LogSinkInner) {
    let mut pending = lock_pending(&inner.pending);
    let due = pending
        .opened_at
        .is_some_and(|opened| opened.elapsed() >= LOG_EVENT_BATCH_INTERVAL);
    if due {
        emit_pending(inner, &mut pending);
    }
}

/// 低输出量时不会触发"攒满即发"，靠这个后台任务按时间窗口把日志推出去。
fn spawn_pending_flusher(inner: &Arc<LogSinkInner>) {
    let weak = Arc::downgrade(inner);
    tokio::spawn(async move {
        loop {
            sleep(LOG_EVENT_BATCH_INTERVAL / 2).await;
            let Some(inner) = weak.upgrade() else {
                break;
            };
            flush_pending_if_due(&inner);
        }
    });
}

fn lock_running(
    running: &Mutex<HashMap<String, RunningHandle>>,
) -> std::sync::MutexGuard<'_, HashMap<String, RunningHandle>> {
    running
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
}

/// 只在句柄仍属于指定 run 时移除；`None` 表示"刚插入、还没拿到 run id"的占位句柄。
fn remove_running_if(
    running: &Mutex<HashMap<String, RunningHandle>>,
    task_id: &str,
    run_id: Option<i64>,
) {
    let mut running = lock_running(running);
    if running
        .get(task_id)
        .is_some_and(|handle| handle.run_id == run_id)
    {
        running.remove(task_id);
    }
}

fn spawn_log_reader<R>(
    reader: R,
    stream: &'static str,
    sink: LogSink,
) -> tokio::task::JoinHandle<()>
where
    R: AsyncRead + Unpin + Send + 'static,
{
    tokio::spawn(async move {
        let mut reader = BufReader::new(reader);
        let mut line = Vec::new();
        loop {
            match read_line_capped(&mut reader, &mut line, MAX_LOG_LINE_BYTES).await {
                Ok((0, _)) => break,
                Ok((_, truncated)) => {
                    let text = sink.decode_output(&line);
                    let mut payload = format!("[{stream}] {text}");
                    if truncated {
                        payload.push_str(&format!(
                            "\n[system] 单行输出超过 {MAX_LOG_LINE_BYTES} 字节，已截断\n"
                        ));
                    }
                    sink.append(&payload).await;
                }
                Err(error) => {
                    tracing::warn!(stream, %error, "failed to read task output");
                    break;
                }
            }
        }
    })
}

/// 读取一行，最多保留 `max` 字节。超出上限时丢弃剩余部分但继续读到换行，
/// 既保住内存上限，也不会让后续输出错位。
async fn read_line_capped<R>(
    reader: &mut R,
    buffer: &mut Vec<u8>,
    max: usize,
) -> std::io::Result<(usize, bool)>
where
    R: AsyncBufRead + Unpin,
{
    buffer.clear();
    let mut truncated = false;
    loop {
        let chunk = reader.fill_buf().await?;
        if chunk.is_empty() {
            return Ok((buffer.len(), truncated));
        }
        let newline = chunk.iter().position(|byte| *byte == b'\n');
        let take = newline.map_or(chunk.len(), |index| index + 1);
        let remaining = max.saturating_sub(buffer.len());
        let copied = take.min(remaining);
        buffer.extend_from_slice(&chunk[..copied]);
        if copied < take {
            truncated = true;
        }
        reader.consume(take);
        if newline.is_some() {
            return Ok((buffer.len(), truncated));
        }
    }
}

async fn wait_after_termination(child: &mut Child) {
    if timeout(Duration::from_secs(TERMINATION_WAIT_SECS), child.wait())
        .await
        .is_err()
    {
        let _ = child.kill().await;
    }
}

fn env_has(env: &BTreeMap<String, String>, key: &str) -> bool {
    env.keys()
        .any(|candidate| candidate.eq_ignore_ascii_case(key))
}

fn sensitive_values(action: &TaskAction) -> Vec<String> {
    match action {
        TaskAction::Script { env, .. } | TaskAction::Process { env, .. } => env
            .iter()
            .filter(|(key, value)| is_sensitive_name(key) && !value.is_empty())
            .map(|(_, value)| value.clone())
            .collect(),
        TaskAction::Http { headers, .. } => headers
            .iter()
            .filter(|(key, value)| is_sensitive_name(key) && !value.is_empty())
            .map(|(_, value)| value.clone())
            .collect(),
    }
}

fn is_sensitive_name(name: &str) -> bool {
    let name = name.to_ascii_uppercase();
    [
        "SECRET",
        "TOKEN",
        "PASSWORD",
        "PASS",
        "API_KEY",
        "PRIVATE_KEY",
        "CREDENTIAL",
        "AUTHORIZATION",
        "AUTH",
    ]
    .iter()
    .any(|marker| name.contains(marker))
}

fn redact_text(text: &str, secrets: &[String]) -> String {
    secrets.iter().fold(text.to_owned(), |current, secret| {
        current.replace(secret, "[REDACTED]")
    })
}

fn failed_outcome(message: impl Into<String>) -> ScriptOutcome {
    ScriptOutcome {
        status: RunStatus::Failed,
        exit_code: None,
        error_message: Some(message.into()),
    }
}

fn cleanup_logs_sync(
    root: &Path,
    config: &LogRetentionConfig,
    active: &std::collections::HashSet<PathBuf>,
) -> std::io::Result<()> {
    if !root.exists() {
        return Ok(());
    }

    let mut files = Vec::new();
    collect_log_files(root, &mut files)?;
    // 正在写入的日志不参与保留期/容量清理。
    files.retain(|(path, _)| !active.contains(path));
    let retention = Duration::from_secs(config.retention_days.saturating_mul(86_400));
    let cutoff = SystemTime::now().checked_sub(retention);

    if let Some(cutoff) = cutoff {
        files.retain(|(path, metadata)| {
            let expired = metadata
                .modified()
                .map(|modified| modified < cutoff)
                .unwrap_or(false);
            if expired {
                let _ = std::fs::remove_file(path);
                false
            } else {
                true
            }
        });
    }

    files.sort_by_key(|(_, metadata)| metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH));
    let mut total = files
        .iter()
        .map(|(_, metadata)| metadata.len())
        .sum::<u64>();
    for (path, metadata) in files {
        if total <= config.max_total_bytes {
            break;
        }
        if std::fs::remove_file(&path).is_ok() {
            total = total.saturating_sub(metadata.len());
        }
    }

    remove_empty_directories(root)?;
    Ok(())
}

fn collect_log_files(
    directory: &Path,
    files: &mut Vec<(PathBuf, std::fs::Metadata)>,
) -> std::io::Result<()> {
    for entry in std::fs::read_dir(directory)? {
        let entry = entry?;
        let file_type = entry.file_type()?;
        if file_type.is_symlink() {
            continue;
        }
        if file_type.is_dir() {
            collect_log_files(&entry.path(), files)?;
        } else if file_type.is_file() {
            files.push((entry.path(), entry.metadata()?));
        }
    }
    Ok(())
}

fn remove_empty_directories(root: &Path) -> std::io::Result<()> {
    for entry in std::fs::read_dir(root)? {
        let entry = entry?;
        if entry.file_type()?.is_dir() {
            let path = entry.path();
            remove_empty_directories(&path)?;
            if std::fs::read_dir(&path)?.next().is_none() {
                let _ = std::fs::remove_dir(path);
            }
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn terminal_statuses_are_identified() {
        assert!(!RunStatus::Queued.is_terminal());
        assert!(!RunStatus::Running.is_terminal());
        assert!(RunStatus::Succeeded.is_terminal());
        assert!(RunStatus::Failed.is_terminal());
        assert!(RunStatus::TimedOut.is_terminal());
        assert!(RunStatus::Cancelled.is_terminal());
        assert!(RunStatus::Skipped.is_terminal());
    }

    #[test]
    fn redacts_sensitive_values() {
        let secrets = vec!["super-secret".to_owned(), "token-value".to_owned()];
        let output = redact_text("found super-secret and token-value", &secrets);

        assert_eq!(output, "found [REDACTED] and [REDACTED]");
    }

    #[test]
    fn identifies_sensitive_environment_names() {
        assert!(is_sensitive_name("API_TOKEN"));
        assert!(is_sensitive_name("DB_PASSWORD"));
        assert!(is_sensitive_name("Authorization"));
        assert!(!is_sensitive_name("PYTHONUNBUFFERED"));
    }

    fn auto_encoding(fallback: &'static Encoding) -> OutputEncoding {
        OutputEncoding {
            forced: None,
            fallback,
        }
    }

    #[test]
    fn decodes_utf8_output_without_touching_it() {
        let encoding = auto_encoding(encoding_rs::GBK);

        assert_eq!(encoding.decode("中文输出".as_bytes()), "中文输出");
    }

    #[test]
    fn decodes_gbk_output_through_the_system_fallback() {
        let encoding = auto_encoding(encoding_rs::GBK);
        // "中文" 的 GBK 编码
        let gbk = [0xD6, 0xD0, 0xCE, 0xC4];

        assert_eq!(encoding.decode(&gbk), "中文");
    }

    #[test]
    fn forced_encoding_skips_utf8_sniffing() {
        let encoding = OutputEncoding {
            forced: Some(encoding_rs::GBK),
            fallback: encoding_rs::UTF_8,
        };
        let gbk = [0xD6, 0xD0, 0xCE, 0xC4];

        assert_eq!(encoding.decode(&gbk), "中文");
    }

    #[test]
    fn undecodable_bytes_still_produce_text() {
        let encoding = auto_encoding(encoding_rs::UTF_8);

        assert!(encoding.decode(&[0xFF, 0xFE, 0xFF]).contains('\u{fffd}'));
    }

    #[test]
    fn maps_windows_code_pages_to_encodings() {
        assert_eq!(code_page_encoding(936).name(), "GBK");
        assert_eq!(code_page_encoding(65001).name(), "UTF-8");
        assert_eq!(code_page_encoding(1252).name(), "windows-1252");
    }

    #[test]
    fn cleanup_enforces_total_log_capacity() {
        let temp = tempfile::TempDir::new().unwrap();
        for index in 0..6 {
            std::fs::write(temp.path().join(format!("run-{index}.log")), vec![b'x'; 4]).unwrap();
            std::thread::sleep(std::time::Duration::from_millis(5));
        }

        cleanup_logs_sync(
            temp.path(),
            &LogRetentionConfig {
                retention_days: 30,
                max_total_bytes: 10,
            },
            &std::collections::HashSet::new(),
        )
        .unwrap();
        let total = std::fs::read_dir(temp.path())
            .unwrap()
            .map(|entry| entry.unwrap().metadata().unwrap().len())
            .sum::<u64>();
        assert!(total <= 10);
    }

    #[test]
    fn cleanup_skips_logs_of_running_tasks() {
        let temp = tempfile::TempDir::new().unwrap();
        let active = temp.path().join("active.log");
        std::fs::write(&active, vec![b'x'; 32]).unwrap();
        std::thread::sleep(std::time::Duration::from_millis(5));
        let stale = temp.path().join("stale.log");
        std::fs::write(&stale, vec![b'x'; 32]).unwrap();

        let mut protected = std::collections::HashSet::new();
        protected.insert(active.clone());
        cleanup_logs_sync(
            temp.path(),
            &LogRetentionConfig {
                retention_days: 0,
                max_total_bytes: 1,
            },
            &protected,
        )
        .unwrap();

        assert!(active.exists(), "active log must survive cleanup");
        assert!(!stale.exists(), "stale log should be removed");
    }

    #[tokio::test]
    async fn capped_line_reader_truncates_but_keeps_reading() {
        let payload = [b'a'; 64];
        let mut reader = BufReader::new(&payload[..]);
        let mut buffer = Vec::new();

        let (bytes, truncated) = read_line_capped(&mut reader, &mut buffer, 16)
            .await
            .unwrap();

        assert_eq!(bytes, 16);
        assert!(truncated);

        let (next, next_truncated) = read_line_capped(&mut reader, &mut buffer, 16)
            .await
            .unwrap();
        assert_eq!(next, 0);
        assert!(!next_truncated);
    }

    #[test]
    fn removes_only_the_matching_run_handle() {
        let running = Mutex::new(HashMap::from([(
            "task".to_owned(),
            RunningHandle {
                run_id: Some(2),
                cancellation: CancellationToken::new(),
            },
        )]));

        remove_running_if(&running, "task", Some(1));
        assert!(lock_running(&running).contains_key("task"));

        remove_running_if(&running, "task", Some(2));
        assert!(!lock_running(&running).contains_key("task"));
    }

    async fn test_sink() -> (
        LogSink,
        broadcast::Receiver<ExecutionEvent>,
        tempfile::TempDir,
        PathBuf,
    ) {
        let temp = tempfile::TempDir::new().unwrap();
        let path = temp.path().join("run.log");
        let (events, receiver) = broadcast::channel(32);
        let sink = LogSink::create(
            &path,
            "task",
            7,
            events,
            Vec::new(),
            OutputEncoding::from_env(),
        )
        .await
        .unwrap();
        (sink, receiver, temp, path)
    }

    async fn recv_log(receiver: &mut broadcast::Receiver<ExecutionEvent>) -> (u64, u64, String) {
        let event = timeout(Duration::from_secs(2), receiver.recv())
            .await
            .expect("log event should arrive")
            .expect("log channel should stay open");
        match event {
            ExecutionEvent::Log {
                run_id,
                offset,
                next_offset,
                content,
            } => {
                assert_eq!(run_id, 7);
                (offset, next_offset, content)
            }
            other => panic!("expected log event, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn coalesces_rapid_output_into_a_single_log_event() {
        let (sink, mut receiver, _temp, path) = test_sink().await;

        sink.append("first line\n").await;
        sink.append("second line\n").await;
        drop(sink);

        let (offset, next_offset, content) = recv_log(&mut receiver).await;
        assert_eq!(content, "first line\nsecond line\n");
        assert_eq!(next_offset - offset, content.len() as u64);
        let on_disk = std::fs::read_to_string(&path).unwrap();
        assert!(on_disk.ends_with(&content));
    }

    #[tokio::test]
    async fn flushes_the_pending_batch_when_the_sink_is_dropped() {
        let (sink, mut receiver, _temp, _path) = test_sink().await;

        sink.append("only line\n").await;
        drop(sink);

        let (offset, next_offset, content) = recv_log(&mut receiver).await;
        assert_eq!(content, "only line\n");
        assert_eq!(next_offset - offset, content.len() as u64);
    }

    #[tokio::test]
    async fn flushes_a_full_batch_without_waiting_for_the_window() {
        let (sink, mut receiver, _temp, _path) = test_sink().await;

        sink.append(&"x".repeat(LOG_EVENT_BATCH_BYTES)).await;

        let (_offset, _next_offset, content) = recv_log(&mut receiver).await;
        assert_eq!(content.len(), LOG_EVENT_BATCH_BYTES);
    }

    #[tokio::test]
    async fn flushes_low_volume_output_within_the_time_window() {
        let (sink, mut receiver, _temp, _path) = test_sink().await;

        sink.append("quiet output\n").await;

        let (_offset, _next_offset, content) = recv_log(&mut receiver).await;
        assert_eq!(content, "quiet output\n");
    }

    #[tokio::test]
    async fn batches_keep_contiguous_offsets_across_flushes() {
        let (sink, mut receiver, _temp, _path) = test_sink().await;

        sink.append(&"x".repeat(LOG_EVENT_BATCH_BYTES)).await;
        let (first_offset, first_next, first_content) = recv_log(&mut receiver).await;
        sink.append("tail\n").await;
        let (second_offset, second_next, second_content) = recv_log(&mut receiver).await;

        assert_eq!(first_next - first_offset, first_content.len() as u64);
        assert_eq!(second_offset, first_next);
        assert_eq!(second_next - second_offset, second_content.len() as u64);
        assert_eq!(second_content, "tail\n");
    }
}