BlueArchiveToolkit/infrastructure/src/translation_tasks.rs

//! Durable translation-task state for official TextUnit handoff.
//!
//! The official TextUnit queue remains an immutable release artifact. This
//! module stores the mutable worker state separately so a provider worker can
//! retry or complete a task without rewriting the published release.

use crate::official_textunit_queue::{
    textunit_task_matches, OfficialTextUnitTask, OfficialTextUnitTaskQuery,
    OfficialTextUnitTaskQueue, OFFICIAL_TEXTUNIT_TASK_QUEUE_FILE,
};
use crate::path_security::{
    ensure_path_within_root, ensure_safe_file_target, read_file_no_symlink, write_file_atomic,
    STATE_FILE_MODE,
};
use bat_core::Result;
use serde::{Deserialize, Serialize};
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
use sqlx::{QueryBuilder, Sqlite, SqlitePool};
use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use std::str::FromStr;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

/// SQLite schema version for durable translation task state.
pub const TRANSLATION_TASK_SCHEMA_VERSION: u32 = 2;
const TRANSLATION_TASK_SCHEMA_COMPONENT: &str = "translation_tasks";
/// SQLite file name stored under a published official release root.
pub const TRANSLATION_TASK_REPOSITORY_FILE: &str = "translation-tasks.sqlite";
/// Translation handoff schema version.
pub const TRANSLATION_HANDOFF_SCHEMA_VERSION: u32 = 1;
/// Versioned handoff file stored under a published official release root.
pub const TRANSLATION_HANDOFF_FILE: &str = "translation-handoff.json";

/// Lifecycle status for one translation job.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TranslationJobStatus {
    /// No provider work has started.
    Queued,
    /// At least one provider run is active.
    Translating,
    /// Provider output is waiting for human review.
    Review,
    /// All source units have translated output ready for patching.
    Ready,
    /// A localized release was published.
    Published,
    /// One or more provider runs failed.
    Failed,
}

impl TranslationJobStatus {
    /// Returns the stable handoff label.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Queued => "queued",
            Self::Translating => "translating",
            Self::Review => "review",
            Self::Ready => "ready",
            Self::Published => "published",
            Self::Failed => "failed",
        }
    }
}

/// Lifecycle status for one resource/TextUnit task in a translation job.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TranslationUnitStatus {
    /// Waiting for a provider.
    Queued,
    /// A provider is processing the unit.
    Translating,
    /// Provider output exists but is not yet reviewed.
    Translated,
    /// Human review accepted the output.
    Reviewed,
    /// A localized patch was generated.
    Patched,
    /// The localized release contains this unit.
    Published,
    /// Provider processing failed.
    Failed,
    /// The source task was skipped by parsing or policy.
    Skipped,
}

/// Status of one provider execution recorded in a translation handoff.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ProviderRunStatus {
    /// Waiting to be scheduled.
    Queued,
    /// Provider work is active.
    Running,
    /// Provider work completed.
    Succeeded,
    /// Provider work failed.
    Failed,
    /// Provider work was cancelled.
    Cancelled,
}

/// Durable job-level translation state.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TranslationJob {
    /// Stable job ID derived from the official release.
    pub job_id: String,
    /// Official release consumed by this job.
    pub official_release_id: String,
    /// Previous official release, when known.
    pub previous_release_id: Option<String>,
    /// Localized release produced by a later patch/publish stage.
    pub localized_release_id: Option<String>,
    /// Current job lifecycle status.
    pub status: TranslationJobStatus,
    /// Number of source units represented by the handoff.
    pub unit_count: usize,
    /// Queue generation time.
    pub created_unix_seconds: u64,
    /// Last handoff state generation time.
    pub updated_unix_seconds: u64,
    /// Job-level diagnostic, if failed.
    pub failure_reason: Option<String>,
}

/// One translation unit linked back to an immutable official task.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TranslationUnit {
    /// Stable unit ID.
    pub unit_id: String,
    /// Owning translation job.
    pub job_id: String,
    /// Immutable official TextUnit task ID.
    pub task_id: String,
    /// Official release containing the source.
    pub official_release_id: String,
    /// Resource destination under the official release.
    pub destination: String,
    /// ZIP/archive entry, when the source was nested.
    pub archive_entry: Option<String>,
    /// Number of extracted TextUnits represented by this task.
    pub text_unit_count: usize,
    /// Extracted TextUnit format labels.
    pub text_unit_formats: Vec<String>,
    /// Current translation lifecycle state.
    pub status: TranslationUnitStatus,
    /// Provider failure diagnostic, when present.
    pub failure_reason: Option<String>,
    /// TextUnit-level provider results associated with this resource task.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub translation_results: Vec<TranslationTaskUnitResult>,
}

/// One provider result linked to an immutable TextUnit.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TranslationTaskUnitResult {
    /// Stable TextUnit ID from `official-textunit-index.json`.
    pub unit_id: String,
    /// Source text submitted to the provider.
    pub source_text: String,
    /// Provider-produced or human-supplied translation.
    pub translated_text: String,
    /// Result source kind.
    #[serde(default)]
    pub source_kind: TranslationTaskResultSourceKind,
    /// Trusted Translation Memory record used for this result, when applicable.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub translation_memory_record_id: Option<String>,
    /// Provider identifier.
    pub provider: String,
    /// Provider run that produced this result.
    pub provider_run_id: String,
    /// Result persistence time.
    pub translated_unix_seconds: u64,
}

/// Source of one persisted TextUnit translation result.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TranslationTaskResultSourceKind {
    /// Result returned by the configured provider.
    #[default]
    Provider,
    /// Result submitted through the manual task update interface.
    Manual,
    /// Result reused from a trusted Translation Memory entry.
    TranslationMemory,
}

impl TranslationTaskResultSourceKind {
    /// Returns the stable JSON label.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Provider => "provider",
            Self::Manual => "manual",
            Self::TranslationMemory => "translation_memory",
        }
    }
}

/// One provider execution associated with one or more translation units.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProviderRun {
    /// Stable provider run ID.
    pub provider_run_id: String,
    /// Owning translation job.
    pub job_id: String,
    /// Provider identifier, reserved for future plugin implementations.
    pub provider: String,
    /// Current provider run status.
    pub status: ProviderRunStatus,
    /// Unit IDs submitted to this run.
    pub unit_ids: Vec<String>,
    /// Number of attempts represented by this run.
    pub attempt_count: u32,
    /// Provider failure diagnostic, when present.
    pub failure_reason: Option<String>,
}

/// Read-only handoff view consumed by translation workers and query clients.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TranslationHandoff {
    /// Handoff schema version.
    pub handoff_version: u32,
    /// Official queue generation time.
    pub generated_unix_seconds: u64,
    /// Stable source queue file name.
    pub source_queue_file: String,
    /// Job-level state.
    pub job: TranslationJob,
    /// Resource/TextUnit task states.
    pub units: Vec<TranslationUnit>,
    /// Provider run states.
    pub provider_runs: Vec<ProviderRun>,
}

/// Builds a translation handoff from immutable queue data and mutable worker
/// state. No provider or network call is made.
pub fn build_translation_handoff(
    queue: &OfficialTextUnitTaskQueue,
    tasks: &[PersistedTranslationTask],
) -> TranslationHandoff {
    let job_id = format!("official-release:{}", queue.official_release_id);
    let persisted = tasks
        .iter()
        .map(|task| (task.task.task_id.as_str(), task))
        .collect::<BTreeMap<_, _>>();
    let mut units = Vec::with_capacity(queue.tasks.len());
    let mut provider_runs = BTreeMap::<String, ProviderRun>::new();

    for task in &queue.tasks {
        let state = persisted
            .get(task.task_id.as_str())
            .copied()
            .map(|task| {
                (
                    task.task_status,
                    task.failure_reason.clone(),
                    task.attempt_count,
                    task.provider_run_id.clone(),
                    task.provider.clone(),
                    task.translation_results.clone(),
                )
            })
            .unwrap_or_else(|| {
                (
                    initial_task_status(task),
                    if task.status == crate::official_textunit_queue::OfficialTextUnitTaskStatus::SkippedParseFailed {
                        task.reason.clone()
                    } else {
                        None
                    },
                    0,
                    None,
                    None,
                    Vec::new(),
                )
            });
        let unit_status = if state.5.is_empty() {
            match state.0 {
                TranslationTaskStatus::Queued => TranslationUnitStatus::Queued,
                TranslationTaskStatus::Running => TranslationUnitStatus::Translating,
                TranslationTaskStatus::Failed => TranslationUnitStatus::Failed,
                TranslationTaskStatus::Completed => TranslationUnitStatus::Translated,
                TranslationTaskStatus::Skipped => TranslationUnitStatus::Skipped,
            }
        } else {
            // A task can retain successful TM hits while the remaining provider
            // units are failed or waiting for retry.
            TranslationUnitStatus::Translated
        };
        let unit = TranslationUnit {
            unit_id: task.task_id.clone(),
            job_id: job_id.clone(),
            task_id: task.task_id.clone(),
            official_release_id: task.official_release_id.clone(),
            destination: task.destination.clone(),
            archive_entry: task.archive_entry.clone(),
            text_unit_count: task.text_unit_count,
            text_unit_formats: task.text_unit_formats.clone(),
            status: unit_status,
            failure_reason: state.1.clone(),
            translation_results: state.5.clone(),
        };
        if let Some(provider_run_id) = state.3 {
            let run = provider_runs
                .entry(provider_run_id.clone())
                .or_insert_with(|| ProviderRun {
                    provider_run_id,
                    job_id: job_id.clone(),
                    provider: state.4.unwrap_or_else(|| "worker".to_string()),
                    status: ProviderRunStatus::Queued,
                    unit_ids: Vec::new(),
                    attempt_count: 0,
                    failure_reason: None,
                });
            run.unit_ids.push(unit.unit_id.clone());
            run.attempt_count = run.attempt_count.max(state.2);
            let candidate_status = match state.0 {
                TranslationTaskStatus::Running => ProviderRunStatus::Running,
                TranslationTaskStatus::Failed => ProviderRunStatus::Failed,
                TranslationTaskStatus::Completed => ProviderRunStatus::Succeeded,
                TranslationTaskStatus::Queued | TranslationTaskStatus::Skipped => run.status,
            };
            if provider_run_status_rank(candidate_status) > provider_run_status_rank(run.status) {
                run.status = candidate_status;
            }
            if state.1.is_some() {
                run.failure_reason = state.1.clone();
            }
        }
        units.push(unit);
    }

    let job_failure_reason = units.iter().find_map(|unit| unit.failure_reason.clone());
    let has_running = units
        .iter()
        .any(|unit| unit.status == TranslationUnitStatus::Translating);
    let candidate_units = units
        .iter()
        .filter(|unit| unit.status != TranslationUnitStatus::Skipped)
        .collect::<Vec<_>>();
    let all_translated = !candidate_units.is_empty()
        && candidate_units
            .iter()
            .all(|unit| unit.status == TranslationUnitStatus::Translated);
    let job_status = if job_failure_reason.is_some() {
        TranslationJobStatus::Failed
    } else if has_running {
        TranslationJobStatus::Translating
    } else if all_translated {
        TranslationJobStatus::Ready
    } else {
        TranslationJobStatus::Queued
    };

    TranslationHandoff {
        handoff_version: TRANSLATION_HANDOFF_SCHEMA_VERSION,
        generated_unix_seconds: queue.generated_unix_seconds,
        source_queue_file: OFFICIAL_TEXTUNIT_TASK_QUEUE_FILE.to_string(),
        job: TranslationJob {
            job_id,
            official_release_id: queue.official_release_id.clone(),
            previous_release_id: queue.previous_release_id.clone(),
            localized_release_id: None,
            status: job_status,
            unit_count: units.len(),
            created_unix_seconds: queue.generated_unix_seconds,
            updated_unix_seconds: unix_seconds_now(),
            failure_reason: job_failure_reason,
        },
        units,
        provider_runs: provider_runs.into_values().collect(),
    }
}

/// Writes a reviewed handoff view under one release root.
pub fn write_translation_handoff_at(
    resource_root: &Path,
    handoff: &TranslationHandoff,
) -> std::result::Result<(), String> {
    let path = resource_root.join(TRANSLATION_HANDOFF_FILE);
    ensure_path_within_root(resource_root, &path)?;
    ensure_safe_file_target(resource_root, &path, "翻译 handoff")?;
    let bytes = serde_json::to_vec_pretty(handoff)
        .map_err(|error| format!("序列化翻译 handoff 失败 {}：{error}", path.display()))?;
    write_file_atomic(&path, &bytes, STATE_FILE_MODE, "翻译 handoff")
}

/// Reads the current translation handoff view from one release root.
pub fn read_translation_handoff_at(
    resource_root: &Path,
) -> std::result::Result<Option<TranslationHandoff>, String> {
    let path = resource_root.join(TRANSLATION_HANDOFF_FILE);
    let Some(bytes) = read_file_no_symlink(&path, "翻译 handoff")? else {
        return Ok(None);
    };
    let handoff: TranslationHandoff = serde_json::from_slice(&bytes)
        .map_err(|error| format!("解析翻译 handoff 失败 {}：{error}", path.display()))?;
    if handoff.handoff_version != TRANSLATION_HANDOFF_SCHEMA_VERSION {
        return Err(format!(
            "不支持的翻译 handoff 版本 {}，文件 {}",
            handoff.handoff_version,
            path.display()
        ));
    }
    Ok(Some(handoff))
}

/// Mutable state owned by a translation worker.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TranslationTaskStatus {
    /// The task is waiting for a provider worker.
    Queued,
    /// A provider worker is processing the task.
    Running,
    /// The provider worker failed the task.
    Failed,
    /// The provider worker completed the task.
    Completed,
    /// The task was intentionally excluded from provider processing.
    Skipped,
}

impl TranslationTaskStatus {
    /// Returns the stable RPC and database label.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Queued => "queued",
            Self::Running => "running",
            Self::Failed => "failed",
            Self::Completed => "completed",
            Self::Skipped => "skipped",
        }
    }

    /// Parses a stable RPC or database label.
    pub fn parse(value: &str) -> Option<Self> {
        match value {
            "queued" => Some(Self::Queued),
            "running" => Some(Self::Running),
            "failed" => Some(Self::Failed),
            "completed" => Some(Self::Completed),
            "skipped" => Some(Self::Skipped),
            _ => None,
        }
    }
}

/// One official TextUnit task with mutable provider state.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PersistedTranslationTask {
    /// Immutable task data derived from the published release.
    #[serde(flatten)]
    pub task: OfficialTextUnitTask,
    /// Current worker state.
    pub task_status: TranslationTaskStatus,
    /// Provider failure reason, if the worker recorded one.
    pub failure_reason: Option<String>,
    /// Number of provider attempts.
    pub attempt_count: u32,
    /// First persistence time as Unix seconds.
    pub created_unix_seconds: u64,
    /// Last state or metadata update time as Unix seconds.
    pub updated_unix_seconds: u64,
    /// Completion time as Unix seconds, if completed.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub completed_unix_seconds: Option<u64>,
    /// Provider-side run identifier, if known.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub provider_run_id: Option<String>,
    /// Provider-produced results keyed by TextUnit ID.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub translation_results: Vec<TranslationTaskUnitResult>,
    /// Provider identifier used by the latest run.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub provider: Option<String>,
    /// Worker currently holding the lease.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub lease_owner: Option<String>,
    /// Lease expiry as Unix seconds.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub lease_expires_unix_seconds: Option<u64>,
    /// Stable failure classification.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub failure_class: Option<String>,
    /// Whether the latest failure can be retried.
    pub failure_retryable: bool,
    /// Earliest Unix time at which a failed task may be claimed again.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub next_attempt_unix_seconds: Option<u64>,
}

/// Mutable persistence metadata associated with an official TextUnit task.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PersistedTranslationTaskState {
    /// Current worker state.
    pub task_status: TranslationTaskStatus,
    /// Provider failure reason, if the worker recorded one.
    pub failure_reason: Option<String>,
    /// Number of provider attempts.
    pub attempt_count: u32,
    /// First persistence time as Unix seconds.
    pub created_unix_seconds: u64,
    /// Last state or metadata update time as Unix seconds.
    pub updated_unix_seconds: u64,
    /// Completion time as Unix seconds, if completed.
    pub completed_unix_seconds: Option<u64>,
    /// Provider-side run identifier, if known.
    pub provider_run_id: Option<String>,
    /// Provider-produced results keyed by TextUnit ID.
    pub translation_results: Vec<TranslationTaskUnitResult>,
    /// Provider identifier used by the latest run.
    pub provider: Option<String>,
    /// Worker currently holding the lease.
    pub lease_owner: Option<String>,
    /// Lease expiry as Unix seconds.
    pub lease_expires_unix_seconds: Option<u64>,
    /// Stable failure classification.
    pub failure_class: Option<String>,
    /// Whether the latest failure can be retried.
    pub failure_retryable: bool,
    /// Earliest Unix time at which a failed task may be claimed again.
    pub next_attempt_unix_seconds: Option<u64>,
}

/// Provider failure data used to atomically release a task lease.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TranslationTaskFailure {
    /// Task being failed.
    pub task_id: String,
    /// Worker that owns the lease.
    pub worker_id: String,
    /// Provider run associated with the lease.
    pub provider_run_id: String,
    /// Stable failure classification.
    pub failure_class: String,
    /// Redacted diagnostic message.
    pub failure_reason: String,
    /// Whether a future attempt may retry the task.
    pub retryable: bool,
    /// Earliest retry time, when retryable.
    pub next_attempt_unix_seconds: Option<u64>,
}

/// Result of synchronizing an immutable release queue into SQLite.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct TranslationTaskSyncReport {
    /// Number of newly inserted task rows.
    pub inserted_count: usize,
    /// Number of existing rows whose immutable task data was refreshed.
    pub refreshed_count: usize,
    /// Number of existing worker states preserved.
    pub preserved_state_count: usize,
    /// Number of stale rows removed because they are absent from the new queue.
    pub removed_count: usize,
}

/// SQLite-backed translation task state repository.
#[derive(Debug, Clone)]
pub struct SqliteTranslationTaskRepository {
    pool: SqlitePool,
}

impl SqliteTranslationTaskRepository {
    /// Opens or creates the database and applies the translation-task schema.
    pub async fn new(path: impl AsRef<Path>) -> Result<Self> {
        Self::open_with(path.as_ref(), true).await
    }

    /// Opens an existing database and applies compatible migrations.
    pub async fn open(path: impl AsRef<Path>) -> Result<Self> {
        Self::open_with(path.as_ref(), false).await
    }

    async fn open_with(path: &Path, create_if_missing: bool) -> Result<Self> {
        if create_if_missing {
            if let Some(parent) = path.parent() {
                tokio::fs::create_dir_all(parent).await?;
            }
        }

        let options = SqliteConnectOptions::from_str(&format!("sqlite://{}", path.display()))
            .map_err(|error| bat_core::Error::Other(error.into()))?
            .create_if_missing(create_if_missing)
            .journal_mode(SqliteJournalMode::Wal)
            .busy_timeout(Duration::from_secs(30));
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(options)
            .await
            .map_err(|error| bat_core::Error::Other(error.into()))?;
        let repository = Self { pool };
        repository.init_schema().await?;
        Ok(repository)
    }

    /// Returns the durable translation-task database path under one release root.
    pub fn repository_path(resource_root: &Path) -> std::path::PathBuf {
        resource_root.join(TRANSLATION_TASK_REPOSITORY_FILE)
    }

    async fn init_schema(&self) -> Result<()> {
        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS schema_migrations (
                component TEXT PRIMARY KEY NOT NULL,
                version INTEGER NOT NULL CHECK(version >= 1)
            )
            "#,
        )
        .execute(&self.pool)
        .await
        .map_err(db_error)?;

        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS translation_tasks (
                task_id TEXT PRIMARY KEY NOT NULL,
                official_release_id TEXT NOT NULL,
                destination TEXT NOT NULL,
                archive_entry TEXT,
                queue_status TEXT NOT NULL,
                queue_reason TEXT,
                parse_status TEXT,
                text_unit_formats_json TEXT NOT NULL DEFAULT '[]',
                task_json TEXT NOT NULL DEFAULT '{}',
                worker_status TEXT NOT NULL DEFAULT 'queued',
                failure_reason TEXT,
                attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(attempt_count >= 0),
                created_unix_seconds INTEGER NOT NULL,
                updated_unix_seconds INTEGER NOT NULL,
                completed_unix_seconds INTEGER,
                provider_run_id TEXT,
                translation_results_json TEXT NOT NULL DEFAULT '[]',
                provider TEXT,
                lease_owner TEXT,
                lease_expires_unix_seconds INTEGER,
                failure_class TEXT,
                failure_retryable INTEGER NOT NULL DEFAULT 0,
                next_attempt_unix_seconds INTEGER
            )
            "#,
        )
        .execute(&self.pool)
        .await
        .map_err(db_error)?;

        // These defaults keep old experimental databases readable while the
        // schema version table records the migration boundary explicitly.
        ensure_column(&self.pool, "translation_tasks", "queue_reason", "TEXT").await?;
        ensure_column(&self.pool, "translation_tasks", "parse_status", "TEXT").await?;
        ensure_column(
            &self.pool,
            "translation_tasks",
            "text_unit_formats_json",
            "TEXT NOT NULL DEFAULT '[]'",
        )
        .await?;
        ensure_column(
            &self.pool,
            "translation_tasks",
            "task_json",
            "TEXT NOT NULL DEFAULT '{}'",
        )
        .await?;
        ensure_column(
            &self.pool,
            "translation_tasks",
            "worker_status",
            "TEXT NOT NULL DEFAULT 'queued'",
        )
        .await?;
        ensure_column(&self.pool, "translation_tasks", "failure_reason", "TEXT").await?;
        ensure_column(
            &self.pool,
            "translation_tasks",
            "attempt_count",
            "INTEGER NOT NULL DEFAULT 0",
        )
        .await?;
        ensure_column(
            &self.pool,
            "translation_tasks",
            "created_unix_seconds",
            "INTEGER NOT NULL DEFAULT 0",
        )
        .await?;
        ensure_column(
            &self.pool,
            "translation_tasks",
            "updated_unix_seconds",
            "INTEGER NOT NULL DEFAULT 0",
        )
        .await?;
        ensure_column(
            &self.pool,
            "translation_tasks",
            "completed_unix_seconds",
            "INTEGER",
        )
        .await?;
        ensure_column(&self.pool, "translation_tasks", "provider_run_id", "TEXT").await?;
        ensure_column(
            &self.pool,
            "translation_tasks",
            "translation_results_json",
            "TEXT NOT NULL DEFAULT '[]'",
        )
        .await?;
        ensure_column(&self.pool, "translation_tasks", "provider", "TEXT").await?;
        ensure_column(&self.pool, "translation_tasks", "lease_owner", "TEXT").await?;
        ensure_column(
            &self.pool,
            "translation_tasks",
            "lease_expires_unix_seconds",
            "INTEGER",
        )
        .await?;
        ensure_column(&self.pool, "translation_tasks", "failure_class", "TEXT").await?;
        ensure_column(
            &self.pool,
            "translation_tasks",
            "failure_retryable",
            "INTEGER NOT NULL DEFAULT 0",
        )
        .await?;
        ensure_column(
            &self.pool,
            "translation_tasks",
            "next_attempt_unix_seconds",
            "INTEGER",
        )
        .await?;

        let current: Option<i64> =
            sqlx::query_scalar("SELECT version FROM schema_migrations WHERE component = ?1")
                .bind(TRANSLATION_TASK_SCHEMA_COMPONENT)
                .fetch_optional(&self.pool)
                .await
                .map_err(db_error)?;
        if current.is_some_and(|version| version > i64::from(TRANSLATION_TASK_SCHEMA_VERSION)) {
            return Err(bat_core::Error::InvalidArgument(format!(
                "不支持的翻译任务 schema 版本：{}",
                current.unwrap_or_default()
            )));
        }
        sqlx::query(
            r#"
            INSERT INTO schema_migrations(component, version)
            VALUES (?1, ?2)
            ON CONFLICT(component) DO UPDATE SET version = excluded.version
            "#,
        )
        .bind(TRANSLATION_TASK_SCHEMA_COMPONENT)
        .bind(i64::from(TRANSLATION_TASK_SCHEMA_VERSION))
        .execute(&self.pool)
        .await
        .map_err(db_error)?;

        Ok(())
    }

    /// Synchronizes one immutable release queue without resetting worker state.
    pub async fn sync_queue(
        &self,
        queue: &OfficialTextUnitTaskQueue,
    ) -> Result<TranslationTaskSyncReport> {
        let mut transaction = self.pool.begin().await.map_err(db_error)?;
        let mut report = TranslationTaskSyncReport::default();
        for task in &queue.tasks {
            let task_json = serde_json::to_string(task)
                .map_err(|error| bat_core::Error::Serialization(error.to_string()))?;
            let existing: Option<ExistingTaskRow> = sqlx::query_as(
                r#"
                SELECT task_json, worker_status, failure_reason, attempt_count,
                       created_unix_seconds, completed_unix_seconds, provider_run_id,
                       translation_results_json, provider, lease_owner,
                       lease_expires_unix_seconds, failure_class, failure_retryable,
                       next_attempt_unix_seconds
                FROM translation_tasks
                WHERE task_id = ?1
                "#,
            )
            .bind(&task.task_id)
            .fetch_optional(&mut *transaction)
            .await
            .map_err(db_error)?;

            let now = unix_seconds_now_i64();
            let initial_status = initial_task_status(task);
            let initial_failure_reason = (initial_status == TranslationTaskStatus::Skipped)
                .then(|| task.reason.clone())
                .flatten();
            let formats = serde_json::to_string(&task.text_unit_formats)
                .map_err(|error| bat_core::Error::Serialization(error.to_string()))?;
            let parse_status = task.parse_status.map(parse_status_label);

            if let Some((
                previous_task_json,
                worker_status,
                failure_reason,
                attempt_count,
                created,
                completed,
                provider_run_id,
                translation_results_json,
                provider,
                lease_owner,
                lease_expires,
                failure_class,
                failure_retryable,
                next_attempt,
            )) = existing
            {
                let immutable_unchanged = previous_task_json == task_json;
                let (
                    status,
                    failure_reason,
                    attempt_count,
                    created,
                    completed,
                    provider_run_id,
                    translation_results_json,
                    provider,
                    lease_owner,
                    lease_expires,
                    failure_class,
                    failure_retryable,
                    next_attempt,
                ) = if immutable_unchanged {
                    (
                        worker_status,
                        failure_reason,
                        attempt_count,
                        created,
                        completed,
                        provider_run_id,
                        translation_results_json,
                        provider,
                        lease_owner,
                        lease_expires,
                        failure_class,
                        failure_retryable,
                        next_attempt,
                    )
                } else {
                    report.refreshed_count += 1;
                    (
                        initial_status.as_str().to_string(),
                        initial_failure_reason.clone(),
                        0_i64,
                        now,
                        None,
                        None,
                        "[]".to_string(),
                        None,
                        None,
                        None,
                        None,
                        0_i64,
                        None,
                    )
                };
                if immutable_unchanged {
                    report.preserved_state_count += 1;
                }
                sqlx::query(
                    r#"
                    UPDATE translation_tasks
                    SET official_release_id = ?2, destination = ?3, archive_entry = ?4,
                        queue_status = ?5, queue_reason = ?6, parse_status = ?7,
                        text_unit_formats_json = ?8, task_json = ?9,
                        worker_status = ?10, failure_reason = ?11, attempt_count = ?12,
                        created_unix_seconds = ?13, updated_unix_seconds = ?14,
                        completed_unix_seconds = ?15, provider_run_id = ?16,
                        translation_results_json = ?17, provider = ?18,
                        lease_owner = ?19, lease_expires_unix_seconds = ?20,
                        failure_class = ?21, failure_retryable = ?22,
                        next_attempt_unix_seconds = ?23
                    WHERE task_id = ?1
                    "#,
                )
                .bind(&task.task_id)
                .bind(&task.official_release_id)
                .bind(&task.destination)
                .bind(&task.archive_entry)
                .bind(task.status.as_str())
                .bind(&task.reason)
                .bind(parse_status)
                .bind(formats)
                .bind(task_json)
                .bind(status)
                .bind(failure_reason)
                .bind(attempt_count)
                .bind(created)
                .bind(now)
                .bind(completed)
                .bind(provider_run_id)
                .bind(translation_results_json)
                .bind(provider)
                .bind(lease_owner)
                .bind(lease_expires)
                .bind(failure_class)
                .bind(failure_retryable)
                .bind(next_attempt)
                .execute(&mut *transaction)
                .await
                .map_err(db_error)?;
            } else {
                report.inserted_count += 1;
                sqlx::query(
                    r#"
                    INSERT INTO translation_tasks (
                        task_id, official_release_id, destination, archive_entry,
                        queue_status, queue_reason, parse_status, text_unit_formats_json,
                        task_json, worker_status, failure_reason, attempt_count,
                        created_unix_seconds, updated_unix_seconds
                    )
                    VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, 0, ?12, ?12)
                    "#,
                )
                .bind(&task.task_id)
                .bind(&task.official_release_id)
                .bind(&task.destination)
                .bind(&task.archive_entry)
                .bind(task.status.as_str())
                .bind(&task.reason)
                .bind(parse_status)
                .bind(formats)
                .bind(task_json)
                .bind(initial_status.as_str())
                .bind(initial_failure_reason)
                .bind(now)
                .execute(&mut *transaction)
                .await
                .map_err(db_error)?;
            }
        }
        let task_ids = queue
            .tasks
            .iter()
            .map(|task| task.task_id.as_str())
            .collect::<BTreeSet<_>>();
        let removed_count = if task_ids.is_empty() {
            sqlx::query("DELETE FROM translation_tasks")
                .execute(&mut *transaction)
                .await
                .map_err(db_error)?
                .rows_affected() as usize
        } else {
            let mut query =
                QueryBuilder::<Sqlite>::new("DELETE FROM translation_tasks WHERE task_id NOT IN (");
            let mut separated = query.separated(", ");
            for task_id in task_ids {
                separated.push_bind(task_id);
            }
            separated.push_unseparated(")");
            query
                .build()
                .execute(&mut *transaction)
                .await
                .map_err(db_error)?
                .rows_affected() as usize
        };
        report.removed_count = removed_count;
        transaction.commit().await.map_err(db_error)?;
        Ok(report)
    }

    /// Returns all persisted tasks matching the queue and worker filters.
    pub async fn list(
        &self,
        query: &OfficialTextUnitTaskQuery,
    ) -> Result<Vec<PersistedTranslationTask>> {
        let rows: Vec<TranslationTaskRow> = sqlx::query_as(
            r#"
            SELECT task_json, worker_status, failure_reason, attempt_count,
                   created_unix_seconds, updated_unix_seconds,
                   completed_unix_seconds, provider_run_id, translation_results_json,
                   provider, lease_owner, lease_expires_unix_seconds, failure_class,
                   failure_retryable, next_attempt_unix_seconds
            FROM translation_tasks
            ORDER BY task_id
            "#,
        )
        .fetch_all(&self.pool)
        .await
        .map_err(db_error)?;
        rows.into_iter()
            .map(PersistedTranslationTask::from_row)
            .collect::<Result<Vec<_>>>()
            .map(|tasks| {
                tasks
                    .into_iter()
                    .filter(|task| matches_query(task, query))
                    .collect()
            })
    }

    /// Returns the number of persisted tasks matching a query.
    pub async fn count(&self, query: &OfficialTextUnitTaskQuery) -> Result<u64> {
        Ok(self.list(query).await?.len() as u64)
    }

    /// Requeues tasks whose worker lease expired.
    pub async fn recover_expired_leases(&self, now_unix_seconds: u64) -> Result<u64> {
        let now = i64::try_from(now_unix_seconds).unwrap_or(i64::MAX);
        let result = sqlx::query(
            r#"
            UPDATE translation_tasks
            SET worker_status = CASE
                    WHEN attempt_count > 0 THEN 'failed'
                    ELSE 'queued'
                END,
                failure_reason = 'worker lease expired; task recovered',
                failure_class = 'lease_expired',
                failure_retryable = 1,
                next_attempt_unix_seconds = ?1,
                updated_unix_seconds = ?1,
                lease_owner = NULL,
                lease_expires_unix_seconds = NULL
            WHERE worker_status = 'running'
              AND lease_expires_unix_seconds IS NOT NULL
              AND lease_expires_unix_seconds <= ?1
            "#,
        )
        .bind(now)
        .execute(&self.pool)
        .await
        .map_err(db_error)?;
        Ok(result.rows_affected())
    }

    /// Claims one queued or retryable failed task and assigns an exclusive
    /// lease to one worker.
    pub async fn claim_next(
        &self,
        worker_id: &str,
        provider: &str,
        lease_seconds: u64,
        max_attempts: u32,
    ) -> Result<Option<PersistedTranslationTask>> {
        if worker_id.trim().is_empty() || provider.trim().is_empty() {
            return Err(bat_core::Error::InvalidArgument(
                "translation worker_id/provider 不能为空".to_string(),
            ));
        }
        if max_attempts == 0 {
            return Err(bat_core::Error::InvalidArgument(
                "translation worker max_attempts 必须大于 0".to_string(),
            ));
        }
        if lease_seconds == 0 {
            return Err(bat_core::Error::InvalidArgument(
                "translation worker lease_seconds 必须大于 0".to_string(),
            ));
        }
        let now = unix_seconds_now_i64();
        let lease_expires = now.saturating_add(i64::try_from(lease_seconds).unwrap_or(i64::MAX));
        let mut transaction = self.pool.begin().await.map_err(db_error)?;
        let row: Option<TranslationTaskRow> = sqlx::query_as(
            r#"
            SELECT task_json, worker_status, failure_reason, attempt_count,
                   created_unix_seconds, updated_unix_seconds,
                   completed_unix_seconds, provider_run_id, translation_results_json,
                   provider, lease_owner, lease_expires_unix_seconds, failure_class,
                   failure_retryable, next_attempt_unix_seconds
            FROM translation_tasks
            WHERE queue_status = 'queued_offline'
              AND (
                    worker_status = 'queued'
                    OR (
                        worker_status = 'failed'
                        AND failure_retryable = 1
                        AND attempt_count < ?1
                        AND (next_attempt_unix_seconds IS NULL
                             OR next_attempt_unix_seconds <= ?2)
                    )
              )
              AND (lease_expires_unix_seconds IS NULL OR lease_expires_unix_seconds <= ?2)
            ORDER BY task_id
            LIMIT 1
            "#,
        )
        .bind(i64::from(max_attempts))
        .bind(now)
        .fetch_optional(&mut *transaction)
        .await
        .map_err(db_error)?;
        let Some(row) = row else {
            transaction.commit().await.map_err(db_error)?;
            return Ok(None);
        };
        let current = PersistedTranslationTask::from_row(row)?;
        let attempt_count = current.attempt_count.saturating_add(1);
        let provider_run_id = format!(
            "{provider}:{}:attempt-{attempt_count}",
            current.task.task_id
        );
        let result = sqlx::query(
            r#"
            UPDATE translation_tasks
            SET worker_status = 'running',
                failure_reason = NULL,
                attempt_count = ?2,
                updated_unix_seconds = ?3,
                completed_unix_seconds = NULL,
                provider_run_id = ?4,
                provider = ?5,
                lease_owner = ?6,
                lease_expires_unix_seconds = ?7,
                failure_class = NULL,
                failure_retryable = 0,
                next_attempt_unix_seconds = NULL
            WHERE task_id = ?1
              AND (
                    worker_status = 'queued'
                    OR (
                        worker_status = 'failed'
                        AND failure_retryable = 1
                        AND attempt_count < ?8
                        AND (next_attempt_unix_seconds IS NULL
                             OR next_attempt_unix_seconds <= ?3)
                    )
              )
              AND queue_status = 'queued_offline'
              AND (lease_expires_unix_seconds IS NULL OR lease_expires_unix_seconds <= ?3)
            "#,
        )
        .bind(&current.task.task_id)
        .bind(i64::from(attempt_count))
        .bind(now)
        .bind(&provider_run_id)
        .bind(provider)
        .bind(worker_id)
        .bind(lease_expires)
        .bind(i64::from(max_attempts))
        .execute(&mut *transaction)
        .await
        .map_err(db_error)?;
        if result.rows_affected() != 1 {
            transaction.rollback().await.map_err(db_error)?;
            return Ok(None);
        }
        transaction.commit().await.map_err(db_error)?;
        self.find(&current.task.task_id).await.map(Some)
    }

    /// Stores an idempotent provider result while the worker still owns its lease.
    pub async fn complete_claim(
        &self,
        task_id: &str,
        worker_id: &str,
        provider_run_id: &str,
        provider: &str,
        translation_results: &[TranslationTaskUnitResult],
    ) -> Result<PersistedTranslationTask> {
        let results_json = serde_json::to_string(translation_results)
            .map_err(|error| bat_core::Error::Serialization(error.to_string()))?;
        let now = unix_seconds_now_i64();
        let result = sqlx::query(
            r#"
            UPDATE translation_tasks
            SET worker_status = 'completed',
                failure_reason = NULL,
                updated_unix_seconds = ?2,
                completed_unix_seconds = ?2,
                provider_run_id = ?3,
                provider = ?4,
                translation_results_json = ?5,
                lease_owner = NULL,
                lease_expires_unix_seconds = NULL,
                failure_class = NULL,
                failure_retryable = 0,
                next_attempt_unix_seconds = NULL
            WHERE task_id = ?1
              AND worker_status = 'running'
              AND lease_owner = ?6
              AND provider_run_id = ?3
            "#,
        )
        .bind(task_id)
        .bind(now)
        .bind(provider_run_id)
        .bind(provider)
        .bind(results_json)
        .bind(worker_id)
        .execute(&self.pool)
        .await
        .map_err(db_error)?;
        if result.rows_affected() != 1 {
            return Err(bat_core::Error::InvalidArgument(format!(
                "翻译任务 {} 的 lease 已失效，拒绝写入 provider 结果",
                task_id
            )));
        }
        self.find(task_id).await
    }

    /// Records one provider failure and releases the worker lease.
    pub async fn fail_claim(
        &self,
        failure: TranslationTaskFailure,
    ) -> Result<PersistedTranslationTask> {
        self.fail_claim_with_results(failure, &[]).await
    }

    /// Records a provider failure while retaining any already-resolved TextUnit
    /// results, such as trusted Translation Memory hits.
    pub async fn fail_claim_with_results(
        &self,
        failure: TranslationTaskFailure,
        translation_results: &[TranslationTaskUnitResult],
    ) -> Result<PersistedTranslationTask> {
        let now = unix_seconds_now_i64();
        let next_attempt = failure
            .next_attempt_unix_seconds
            .map(|value| i64::try_from(value).unwrap_or(i64::MAX));
        let translation_results_json = if translation_results.is_empty() {
            None
        } else {
            Some(
                serde_json::to_string(translation_results)
                    .map_err(|error| bat_core::Error::Serialization(error.to_string()))?,
            )
        };
        let result = sqlx::query(
            r#"
            UPDATE translation_tasks
            SET worker_status = 'failed',
                failure_reason = ?2,
                updated_unix_seconds = ?3,
                lease_owner = NULL,
                lease_expires_unix_seconds = NULL,
                failure_class = ?4,
                failure_retryable = ?5,
                next_attempt_unix_seconds = ?6,
                translation_results_json = COALESCE(?7, translation_results_json)
            WHERE task_id = ?1
              AND worker_status = 'running'
              AND lease_owner = ?8
              AND provider_run_id = ?9
            "#,
        )
        .bind(&failure.task_id)
        .bind(&failure.failure_reason)
        .bind(now)
        .bind(&failure.failure_class)
        .bind(if failure.retryable { 1_i64 } else { 0_i64 })
        .bind(next_attempt)
        .bind(translation_results_json)
        .bind(&failure.worker_id)
        .bind(&failure.provider_run_id)
        .execute(&self.pool)
        .await
        .map_err(db_error)?;
        if result.rows_affected() != 1 {
            return Err(bat_core::Error::InvalidArgument(format!(
                "翻译任务 {} 的 lease 已失效，拒绝写入 provider 失败状态",
                failure.task_id
            )));
        }
        self.find(&failure.task_id).await
    }

    /// Updates provider state and returns the durable task record.
    pub async fn update_status(
        &self,
        task_id: &str,
        status: TranslationTaskStatus,
        failure_reason: Option<String>,
        provider_run_id: Option<String>,
    ) -> Result<PersistedTranslationTask> {
        self.update_status_with_results(
            task_id,
            status,
            failure_reason,
            provider_run_id,
            None,
            None,
        )
        .await
    }

    /// Updates provider state and optionally replaces durable TextUnit results.
    pub async fn update_status_with_results(
        &self,
        task_id: &str,
        status: TranslationTaskStatus,
        failure_reason: Option<String>,
        provider_run_id: Option<String>,
        provider: Option<String>,
        translation_results: Option<&[TranslationTaskUnitResult]>,
    ) -> Result<PersistedTranslationTask> {
        let current = self.find(task_id).await?;
        let now = unix_seconds_now_i64();
        let attempt_count = if status == TranslationTaskStatus::Running
            && current.task_status != TranslationTaskStatus::Running
        {
            current.attempt_count.saturating_add(1)
        } else {
            current.attempt_count
        };
        let normalized_reason = failure_reason.filter(|reason| !reason.trim().is_empty());
        let provider_run_id = provider_run_id.filter(|value| !value.trim().is_empty());
        let provider = provider.filter(|value| !value.trim().is_empty());
        let translation_results_json =
            translation_results
                .map(serde_json::to_string)
                .transpose()
                .map_err(|error| bat_core::Error::Serialization(error.to_string()))?;
        let completed = (status == TranslationTaskStatus::Completed).then_some(now);
        sqlx::query(
            r#"
            UPDATE translation_tasks
            SET worker_status = ?2, failure_reason = ?3, attempt_count = ?4,
                updated_unix_seconds = ?5, completed_unix_seconds = ?6,
                provider_run_id = COALESCE(?7, provider_run_id),
                provider = COALESCE(?8, provider),
                translation_results_json = COALESCE(?9, translation_results_json),
                lease_owner = NULL, lease_expires_unix_seconds = NULL,
                failure_class = NULL, failure_retryable = 0,
                next_attempt_unix_seconds = NULL
            WHERE task_id = ?1
            "#,
        )
        .bind(task_id)
        .bind(status.as_str())
        .bind(normalized_reason)
        .bind(i64::from(attempt_count))
        .bind(now)
        .bind(completed)
        .bind(provider_run_id)
        .bind(provider)
        .bind(translation_results_json)
        .execute(&self.pool)
        .await
        .map_err(db_error)?;
        self.find(task_id).await
    }

    /// Finds one task by its stable ID.
    pub async fn find(&self, task_id: &str) -> Result<PersistedTranslationTask> {
        let row: Option<TranslationTaskRow> = sqlx::query_as(
            r#"
            SELECT task_json, worker_status, failure_reason, attempt_count,
                   created_unix_seconds, updated_unix_seconds,
                   completed_unix_seconds, provider_run_id, translation_results_json,
                   provider, lease_owner, lease_expires_unix_seconds, failure_class,
                   failure_retryable, next_attempt_unix_seconds
            FROM translation_tasks
            WHERE task_id = ?1
            "#,
        )
        .bind(task_id)
        .fetch_optional(&self.pool)
        .await
        .map_err(db_error)?;
        row.map(PersistedTranslationTask::from_row)
            .transpose()?
            .ok_or_else(|| bat_core::Error::NotFound(task_id.to_string()))
    }
}

type ExistingTaskRow = (
    String,
    String,
    Option<String>,
    i64,
    i64,
    Option<i64>,
    Option<String>,
    String,
    Option<String>,
    Option<String>,
    Option<i64>,
    Option<String>,
    i64,
    Option<i64>,
);

type TranslationTaskRow = (
    String,
    String,
    Option<String>,
    i64,
    i64,
    i64,
    Option<i64>,
    Option<String>,
    String,
    Option<String>,
    Option<String>,
    Option<i64>,
    Option<String>,
    i64,
    Option<i64>,
);

impl PersistedTranslationTask {
    /// Builds a persisted translation task from an immutable queue task.
    pub fn from_task(task: OfficialTextUnitTask, state: PersistedTranslationTaskState) -> Self {
        Self {
            task,
            task_status: state.task_status,
            failure_reason: state.failure_reason,
            attempt_count: state.attempt_count,
            created_unix_seconds: state.created_unix_seconds,
            updated_unix_seconds: state.updated_unix_seconds,
            completed_unix_seconds: state.completed_unix_seconds,
            provider_run_id: state.provider_run_id,
            translation_results: state.translation_results,
            provider: state.provider,
            lease_owner: state.lease_owner,
            lease_expires_unix_seconds: state.lease_expires_unix_seconds,
            failure_class: state.failure_class,
            failure_retryable: state.failure_retryable,
            next_attempt_unix_seconds: state.next_attempt_unix_seconds,
        }
    }

    /// Builds a synthetic persisted task for older release roots without SQLite state.
    pub fn from_queued_task(task: OfficialTextUnitTask, generated_unix_seconds: u64) -> Self {
        let task_status = initial_task_status(&task);
        let failure_reason = if task_status == TranslationTaskStatus::Skipped {
            task.reason.clone()
        } else {
            None
        };
        Self::from_task(
            task,
            PersistedTranslationTaskState {
                task_status,
                failure_reason,
                attempt_count: 0,
                created_unix_seconds: generated_unix_seconds,
                updated_unix_seconds: generated_unix_seconds,
                completed_unix_seconds: None,
                provider_run_id: None,
                translation_results: Vec::new(),
                provider: None,
                lease_owner: None,
                lease_expires_unix_seconds: None,
                failure_class: None,
                failure_retryable: false,
                next_attempt_unix_seconds: None,
            },
        )
    }

    fn from_row(row: TranslationTaskRow) -> Result<Self> {
        let (
            task_json,
            worker_status,
            failure_reason,
            attempt_count,
            created_unix_seconds,
            updated_unix_seconds,
            completed_unix_seconds,
            provider_run_id,
            translation_results_json,
            provider,
            lease_owner,
            lease_expires_unix_seconds,
            failure_class,
            failure_retryable,
            next_attempt_unix_seconds,
        ) = row;
        let task = serde_json::from_str(&task_json)
            .map_err(|error| bat_core::Error::Serialization(error.to_string()))?;
        let task_status = TranslationTaskStatus::parse(&worker_status).ok_or_else(|| {
            bat_core::Error::Serialization(format!("未知翻译任务 worker 状态：{worker_status}"))
        })?;
        let translation_results = serde_json::from_str(&translation_results_json)
            .map_err(|error| bat_core::Error::Serialization(error.to_string()))?;
        Ok(Self {
            task,
            task_status,
            failure_reason,
            attempt_count: u32::try_from(attempt_count).map_err(|_| {
                bat_core::Error::Serialization("翻译任务 attempt_count 超出范围".to_string())
            })?,
            created_unix_seconds: u64::try_from(created_unix_seconds).map_err(|_| {
                bat_core::Error::Serialization("翻译任务 created 时间无效".to_string())
            })?,
            updated_unix_seconds: u64::try_from(updated_unix_seconds).map_err(|_| {
                bat_core::Error::Serialization("翻译任务 updated 时间无效".to_string())
            })?,
            completed_unix_seconds: completed_unix_seconds
                .map(|value| {
                    u64::try_from(value).map_err(|_| {
                        bat_core::Error::Serialization("翻译任务 completed 时间无效".to_string())
                    })
                })
                .transpose()?,
            provider_run_id,
            translation_results,
            provider,
            lease_owner,
            lease_expires_unix_seconds: lease_expires_unix_seconds
                .map(|value| {
                    u64::try_from(value).map_err(|_| {
                        bat_core::Error::Serialization("翻译任务 lease 时间无效".to_string())
                    })
                })
                .transpose()?,
            failure_class,
            failure_retryable: failure_retryable != 0,
            next_attempt_unix_seconds: next_attempt_unix_seconds
                .map(|value| {
                    u64::try_from(value).map_err(|_| {
                        bat_core::Error::Serialization("翻译任务 next_attempt 时间无效".to_string())
                    })
                })
                .transpose()?,
        })
    }
}

/// Synchronizes a release queue into the durable SQLite repository for that release.
pub async fn sync_translation_task_repository_at(
    resource_root: &Path,
    queue: &OfficialTextUnitTaskQueue,
) -> Result<TranslationTaskSyncReport> {
    let path = SqliteTranslationTaskRepository::repository_path(resource_root);
    let repository = SqliteTranslationTaskRepository::new(&path).await?;
    repository.sync_queue(queue).await
}

fn matches_query(task: &PersistedTranslationTask, query: &OfficialTextUnitTaskQuery) -> bool {
    let mut queue_query = query.clone();
    queue_query.task_status = None;
    if !textunit_task_matches(&task.task, &queue_query) {
        return false;
    }
    if query
        .task_status
        .as_ref()
        .is_some_and(|status| task.task_status.as_str() != status)
    {
        return false;
    }
    if let Some(has_failure_reason) = query.has_failure_reason {
        if task.failure_reason.is_some() != has_failure_reason {
            return false;
        }
    }
    true
}

fn initial_task_status(task: &OfficialTextUnitTask) -> TranslationTaskStatus {
    match task.status {
        crate::official_textunit_queue::OfficialTextUnitTaskStatus::QueuedOffline => {
            TranslationTaskStatus::Queued
        }
        crate::official_textunit_queue::OfficialTextUnitTaskStatus::SkippedNoParseEntry
        | crate::official_textunit_queue::OfficialTextUnitTaskStatus::SkippedNoTextUnit
        | crate::official_textunit_queue::OfficialTextUnitTaskStatus::SkippedParseFailed
        | crate::official_textunit_queue::OfficialTextUnitTaskStatus::SkippedUnsupported => {
            TranslationTaskStatus::Skipped
        }
    }
}

fn parse_status_label(status: crate::official_parse::OfficialParseStatus) -> &'static str {
    match status {
        crate::official_parse::OfficialParseStatus::Parsed => "parsed",
        crate::official_parse::OfficialParseStatus::SkippedUnsupported => "skipped_unsupported",
        crate::official_parse::OfficialParseStatus::Failed => "failed",
    }
}

async fn ensure_column(
    pool: &SqlitePool,
    table: &str,
    column: &str,
    column_type: &str,
) -> Result<()> {
    let exists: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM pragma_table_info(?1) WHERE name = ?2")
            .bind(table)
            .bind(column)
            .fetch_one(pool)
            .await
            .map_err(db_error)?;
    if exists == 0 {
        let mut query = QueryBuilder::<Sqlite>::new("ALTER TABLE ");
        query
            .push(table)
            .push(" ADD COLUMN ")
            .push(column)
            .push(" ");
        query.push(column_type);
        query.build().execute(pool).await.map_err(db_error)?;
    }
    Ok(())
}

fn db_error(error: sqlx::Error) -> bat_core::Error {
    bat_core::Error::Other(error.into())
}

fn unix_seconds_now_i64() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
        .try_into()
        .unwrap_or(i64::MAX)
}

fn unix_seconds_now() -> u64 {
    unix_seconds_now_i64().max(0) as u64
}

fn provider_run_status_rank(status: ProviderRunStatus) -> u8 {
    match status {
        ProviderRunStatus::Queued => 0,
        ProviderRunStatus::Succeeded => 1,
        ProviderRunStatus::Running => 2,
        ProviderRunStatus::Cancelled => 3,
        ProviderRunStatus::Failed => 4,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::official_changes::OfficialResourceChangeKind;
    use crate::official_parse::{OfficialParseSourceKind, OfficialParseStatus};
    use crate::official_textunit_queue::{
        OfficialTextUnitTaskStatus, OfficialTextUnitTaskSummary,
        OFFICIAL_TEXTUNIT_TASK_QUEUE_VERSION,
    };

    fn task(
        task_id: &str,
        destination: &str,
        status: OfficialTextUnitTaskStatus,
        parse_status: Option<OfficialParseStatus>,
        reason: Option<&str>,
    ) -> OfficialTextUnitTask {
        OfficialTextUnitTask {
            task_id: task_id.to_string(),
            official_release_id: "release-current".to_string(),
            destination: destination.to_string(),
            change_kind: OfficialResourceChangeKind::Added,
            url: format!("https://example.invalid/{destination}"),
            bytes: 10,
            blake3: format!("{task_id}-hash"),
            parse_entry_key: Some(format!("direct:{destination}")),
            archive_entry: None,
            source_kind: Some(OfficialParseSourceKind::DirectBundle),
            parse_status,
            text_asset_count: usize::from(status == OfficialTextUnitTaskStatus::QueuedOffline),
            text_assets: if status == OfficialTextUnitTaskStatus::QueuedOffline {
                vec!["Scenario".to_string()]
            } else {
                Vec::new()
            },
            text_unit_count: if status == OfficialTextUnitTaskStatus::QueuedOffline {
                3
            } else {
                0
            },
            text_unit_formats: if status == OfficialTextUnitTaskStatus::QueuedOffline {
                vec!["plain".to_string()]
            } else {
                Vec::new()
            },
            text_unit_error_count: 0,
            status,
            reason: reason.map(str::to_string),
        }
    }

    fn queue(tasks: Vec<OfficialTextUnitTask>) -> OfficialTextUnitTaskQueue {
        OfficialTextUnitTaskQueue {
            queue_version: OFFICIAL_TEXTUNIT_TASK_QUEUE_VERSION,
            official_release_id: "release-current".to_string(),
            previous_release_id: Some("release-previous".to_string()),
            generated_unix_seconds: 123,
            current_resource_root: std::path::PathBuf::from("/tmp/release-current"),
            summary: OfficialTextUnitTaskSummary {
                resource_candidate_count: tasks.len(),
                parse_entry_count: tasks.len(),
                queued_task_count: tasks
                    .iter()
                    .filter(|task| task.status == OfficialTextUnitTaskStatus::QueuedOffline)
                    .count(),
                skipped_parse_failed_count: tasks
                    .iter()
                    .filter(|task| task.status == OfficialTextUnitTaskStatus::SkippedParseFailed)
                    .count(),
                text_unit_count: tasks.iter().map(|task| task.text_unit_count).sum(),
                ..OfficialTextUnitTaskSummary::default()
            },
            tasks,
        }
    }

    #[tokio::test]
    async fn sqlite_translation_tasks_sync_and_preserve_worker_state() {
        let temp = tempfile::TempDir::new().unwrap();
        let repository =
            SqliteTranslationTaskRepository::new(temp.path().join("translation-tasks.sqlite"))
                .await
                .unwrap();
        let queue = queue(vec![
            task(
                "task-a",
                "Bundles/a.bundle",
                OfficialTextUnitTaskStatus::QueuedOffline,
                Some(OfficialParseStatus::Parsed),
                None,
            ),
            task(
                "task-b",
                "Bundles/b.bundle",
                OfficialTextUnitTaskStatus::SkippedParseFailed,
                Some(OfficialParseStatus::Failed),
                Some("parser failed"),
            ),
        ]);

        let first = repository.sync_queue(&queue).await.unwrap();
        assert_eq!(first.inserted_count, 2);
        assert_eq!(first.preserved_state_count, 0);
        let failed = repository
            .list(&OfficialTextUnitTaskQuery {
                task_status: Some("skipped".to_string()),
                has_reason: Some(true),
                ..OfficialTextUnitTaskQuery::default()
            })
            .await
            .unwrap();
        assert_eq!(failed.len(), 1);
        assert_eq!(failed[0].task.task_id, "task-b");
        assert_eq!(failed[0].failure_reason.as_deref(), Some("parser failed"));

        let running = repository
            .update_status(
                "task-a",
                TranslationTaskStatus::Running,
                None,
                Some("run-1".to_string()),
            )
            .await
            .unwrap();
        assert_eq!(running.task_status, TranslationTaskStatus::Running);
        assert_eq!(running.attempt_count, 1);
        let failed = repository
            .update_status(
                "task-a",
                TranslationTaskStatus::Failed,
                Some("remote provider rejected payload".to_string()),
                None,
            )
            .await
            .unwrap();
        assert_eq!(failed.task_status, TranslationTaskStatus::Failed);
        assert_eq!(
            failed.failure_reason.as_deref(),
            Some("remote provider rejected payload")
        );
        assert_eq!(failed.provider_run_id.as_deref(), Some("run-1"));

        let second = repository.sync_queue(&queue).await.unwrap();
        assert_eq!(second.inserted_count, 0);
        assert_eq!(second.preserved_state_count, 2);
        let retrievable = repository
            .list(&OfficialTextUnitTaskQuery {
                task_status: Some("failed".to_string()),
                has_failure_reason: Some(true),
                ..OfficialTextUnitTaskQuery::default()
            })
            .await
            .unwrap();
        assert_eq!(retrievable.len(), 1);
        assert_eq!(retrievable[0].task.task_id, "task-a");
        assert_eq!(retrievable[0].attempt_count, 1);
    }

    #[tokio::test]
    async fn sqlite_translation_tasks_persist_manual_results_without_worker_lease() {
        let temp = tempfile::TempDir::new().unwrap();
        let repository =
            SqliteTranslationTaskRepository::new(temp.path().join("translation-tasks.sqlite"))
                .await
                .unwrap();
        let queue = queue(vec![task(
            "task-a",
            "Bundles/a.bundle",
            OfficialTextUnitTaskStatus::QueuedOffline,
            Some(OfficialParseStatus::Parsed),
            None,
        )]);
        repository.sync_queue(&queue).await.unwrap();
        let result = TranslationTaskUnitResult {
            unit_id: "unit-a".to_string(),
            source_text: "source".to_string(),
            translated_text: "manual translation".to_string(),
            source_kind: TranslationTaskResultSourceKind::Manual,
            translation_memory_record_id: None,
            provider: "manual".to_string(),
            provider_run_id: "manual-run-1".to_string(),
            translated_unix_seconds: 321,
        };

        let updated = repository
            .update_status_with_results(
                "task-a",
                TranslationTaskStatus::Completed,
                None,
                Some("manual-run-1".to_string()),
                Some("manual".to_string()),
                Some(std::slice::from_ref(&result)),
            )
            .await
            .unwrap();

        assert_eq!(updated.task_status, TranslationTaskStatus::Completed);
        assert_eq!(updated.provider.as_deref(), Some("manual"));
        assert_eq!(updated.provider_run_id.as_deref(), Some("manual-run-1"));
        assert_eq!(updated.translation_results, vec![result.clone()]);
        assert_eq!(
            repository.find("task-a").await.unwrap().translation_results,
            vec![result]
        );
    }

    #[tokio::test]
    async fn sqlite_translation_tasks_recover_expired_leases_for_retry() {
        let temp = tempfile::TempDir::new().unwrap();
        let repository =
            SqliteTranslationTaskRepository::new(temp.path().join("translation-tasks.sqlite"))
                .await
                .unwrap();
        let queue = queue(vec![task(
            "task-a",
            "Bundles/a.bundle",
            OfficialTextUnitTaskStatus::QueuedOffline,
            Some(OfficialParseStatus::Parsed),
            None,
        )]);
        repository.sync_queue(&queue).await.unwrap();

        let claimed = repository
            .claim_next("worker-a", "mock", 1, 3)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(claimed.task_status, TranslationTaskStatus::Running);
        assert_eq!(claimed.lease_owner.as_deref(), Some("worker-a"));
        force_expire_lease(&repository, "task-a").await;

        let recovered = repository
            .recover_expired_leases(unix_seconds_now())
            .await
            .unwrap();
        assert_eq!(recovered, 1);
        let recovered_task = repository.find("task-a").await.unwrap();
        assert_eq!(recovered_task.task_status, TranslationTaskStatus::Failed);
        assert_eq!(
            recovered_task.failure_reason.as_deref(),
            Some("worker lease expired; task recovered")
        );
        assert_eq!(
            recovered_task.failure_class.as_deref(),
            Some("lease_expired")
        );
        assert!(recovered_task.failure_retryable);
        assert_eq!(recovered_task.lease_owner, None);

        let reclaimed = repository
            .claim_next("worker-b", "mock", 30, 3)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(reclaimed.task_status, TranslationTaskStatus::Running);
        assert_eq!(reclaimed.attempt_count, 2);
        assert_eq!(reclaimed.lease_owner.as_deref(), Some("worker-b"));
    }

    #[tokio::test]
    async fn sqlite_translation_tasks_reject_stale_worker_writes() {
        let temp = tempfile::TempDir::new().unwrap();
        let repository =
            SqliteTranslationTaskRepository::new(temp.path().join("translation-tasks.sqlite"))
                .await
                .unwrap();
        let queue = queue(vec![task(
            "task-a",
            "Bundles/a.bundle",
            OfficialTextUnitTaskStatus::QueuedOffline,
            Some(OfficialParseStatus::Parsed),
            None,
        )]);
        repository.sync_queue(&queue).await.unwrap();

        let first_claim = repository
            .claim_next("worker-a", "mock", 1, 3)
            .await
            .unwrap()
            .unwrap();
        let first_run = first_claim.provider_run_id.clone().unwrap();
        force_expire_lease(&repository, "task-a").await;
        repository
            .recover_expired_leases(unix_seconds_now())
            .await
            .unwrap();
        let second_claim = repository
            .claim_next("worker-b", "mock", 30, 3)
            .await
            .unwrap()
            .unwrap();
        let second_run = second_claim.provider_run_id.clone().unwrap();
        let result = TranslationTaskUnitResult {
            unit_id: "unit-a".to_string(),
            source_text: "source".to_string(),
            translated_text: "translated".to_string(),
            source_kind: TranslationTaskResultSourceKind::Provider,
            translation_memory_record_id: None,
            provider: "mock".to_string(),
            provider_run_id: second_run.clone(),
            translated_unix_seconds: 1,
        };

        assert!(repository
            .complete_claim(
                "task-a",
                "worker-a",
                &first_run,
                "mock",
                std::slice::from_ref(&result)
            )
            .await
            .is_err());
        let completed = repository
            .complete_claim(
                "task-a",
                "worker-b",
                &second_run,
                "mock",
                std::slice::from_ref(&result),
            )
            .await
            .unwrap();
        assert_eq!(completed.task_status, TranslationTaskStatus::Completed);
        assert_eq!(completed.translation_results, vec![result.clone()]);
        assert!(repository
            .fail_claim(TranslationTaskFailure {
                task_id: "task-a".to_string(),
                worker_id: "worker-a".to_string(),
                provider_run_id: first_run,
                failure_class: "network".to_string(),
                failure_reason: "late failure".to_string(),
                retryable: true,
                next_attempt_unix_seconds: Some(2),
            })
            .await
            .is_err());
        let task = repository.find("task-a").await.unwrap();
        assert_eq!(task.task_status, TranslationTaskStatus::Completed);
        assert_eq!(task.translation_results, vec![result]);
    }

    #[test]
    fn translation_handoff_maps_worker_state_and_preserves_provider_progress() {
        let queue = queue(vec![
            task(
                "task-a",
                "Bundles/a.bundle",
                OfficialTextUnitTaskStatus::QueuedOffline,
                Some(OfficialParseStatus::Parsed),
                None,
            ),
            task(
                "task-b",
                "Bundles/b.bundle",
                OfficialTextUnitTaskStatus::QueuedOffline,
                Some(OfficialParseStatus::Parsed),
                None,
            ),
            task(
                "task-c",
                "Bundles/c.bundle",
                OfficialTextUnitTaskStatus::SkippedParseFailed,
                Some(OfficialParseStatus::Failed),
                Some("parser failed"),
            ),
        ]);
        let persisted = vec![
            PersistedTranslationTask {
                task: queue.tasks[0].clone(),
                task_status: TranslationTaskStatus::Running,
                failure_reason: None,
                attempt_count: 2,
                created_unix_seconds: 123,
                updated_unix_seconds: 124,
                completed_unix_seconds: None,
                provider_run_id: Some("provider-run-1".to_string()),
                translation_results: Vec::new(),
                provider: Some("fixture".to_string()),
                lease_owner: None,
                lease_expires_unix_seconds: None,
                failure_class: None,
                failure_retryable: false,
                next_attempt_unix_seconds: None,
            },
            PersistedTranslationTask {
                task: queue.tasks[1].clone(),
                task_status: TranslationTaskStatus::Completed,
                failure_reason: None,
                attempt_count: 1,
                created_unix_seconds: 123,
                updated_unix_seconds: 125,
                completed_unix_seconds: Some(125),
                provider_run_id: Some("provider-run-1".to_string()),
                translation_results: Vec::new(),
                provider: Some("fixture".to_string()),
                lease_owner: None,
                lease_expires_unix_seconds: None,
                failure_class: None,
                failure_retryable: false,
                next_attempt_unix_seconds: None,
            },
        ];

        let handoff = build_translation_handoff(&queue, &persisted);

        assert_eq!(handoff.handoff_version, TRANSLATION_HANDOFF_SCHEMA_VERSION);
        assert_eq!(handoff.source_queue_file, OFFICIAL_TEXTUNIT_TASK_QUEUE_FILE);
        assert_eq!(handoff.job.status, TranslationJobStatus::Failed);
        assert_eq!(handoff.job.unit_count, 3);
        assert_eq!(handoff.units[0].status, TranslationUnitStatus::Translating);
        assert_eq!(handoff.units[1].status, TranslationUnitStatus::Translated);
        assert_eq!(handoff.units[2].status, TranslationUnitStatus::Skipped);
        assert_eq!(
            handoff.units[2].failure_reason.as_deref(),
            Some("parser failed")
        );
        assert_eq!(handoff.provider_runs.len(), 1);
        assert_eq!(handoff.provider_runs[0].status, ProviderRunStatus::Running);
        assert_eq!(handoff.provider_runs[0].unit_ids, ["task-a", "task-b"]);
        assert_eq!(handoff.provider_runs[0].attempt_count, 2);
    }

    async fn force_expire_lease(repository: &SqliteTranslationTaskRepository, task_id: &str) {
        sqlx::query(
            r#"
            UPDATE translation_tasks
            SET lease_expires_unix_seconds = 0
            WHERE task_id = ?1
            "#,
        )
        .bind(task_id)
        .execute(&repository.pool)
        .await
        .unwrap();
    }

    #[test]
    fn translation_handoff_file_round_trips_with_version_check() {
        let temp = tempfile::TempDir::new().unwrap();
        let handoff = build_translation_handoff(&queue(Vec::new()), &[]);

        write_translation_handoff_at(temp.path(), &handoff).unwrap();
        let loaded = read_translation_handoff_at(temp.path()).unwrap().unwrap();

        assert_eq!(loaded, handoff);
        assert!(temp.path().join(TRANSLATION_HANDOFF_FILE).is_file());
    }
}