feat(i18n): 接入翻译 provider worker

Closes #44

infrastructure/src/translation_tasks.rs

@@ -14,15 +14,15 @@ use crate::path_security::{
 };
 use bat_core::Result;
 use serde::{Deserialize, Serialize};
-use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
+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::{SystemTime, UNIX_EPOCH};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 /// SQLite schema version for durable translation task state.
-pub const TRANSLATION_TASK_SCHEMA_VERSION: u32 = 1;
+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";
@@ -147,6 +147,26 @@ pub struct TranslationUnit {
     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,
+    /// Provider identifier.
+    pub provider: String,
+    /// Provider run that produced this result.
+    pub provider_run_id: String,
+    /// Result persistence time.
+    pub translated_unix_seconds: u64,
 }
 
 /// One provider execution associated with one or more translation units.
@@ -203,7 +223,16 @@ pub fn build_translation_handoff(
         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()))
+            .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),
@@ -214,6 +243,8 @@ pub fn build_translation_handoff(
                     },
                     0,
                     None,
+                    None,
+                    Vec::new(),
                 )
             });
         let unit_status = match state.0 {
@@ -234,6 +265,7 @@ pub fn build_translation_handoff(
             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
@@ -241,7 +273,7 @@ pub fn build_translation_handoff(
                 .or_insert_with(|| ProviderRun {
                     provider_run_id,
                     job_id: job_id.clone(),
-                    provider: "worker".to_string(),
+                    provider: state.4.unwrap_or_else(|| "worker".to_string()),
                     status: ProviderRunStatus::Queued,
                     unit_ids: Vec::new(),
                     attempt_count: 0,
@@ -403,6 +435,26 @@ pub struct PersistedTranslationTask {
     /// 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.
@@ -422,6 +474,39 @@ pub struct PersistedTranslationTaskState {
     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.
@@ -463,7 +548,9 @@ impl SqliteTranslationTaskRepository {
 
         let options = SqliteConnectOptions::from_str(&format!("sqlite://{}", path.display()))
             .map_err(|error| bat_core::Error::Other(error.into()))?
-            .create_if_missing(create_if_missing);
+            .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)
@@ -510,7 +597,14 @@ impl SqliteTranslationTaskRepository {
                 created_unix_seconds INTEGER NOT NULL,
                 updated_unix_seconds INTEGER NOT NULL,
                 completed_unix_seconds INTEGER,
-                provider_run_id TEXT
+                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
             )
             "#,
         )
@@ -573,6 +667,37 @@ impl SqliteTranslationTaskRepository {
         )
         .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")
@@ -615,7 +740,10 @@ impl SqliteTranslationTaskRepository {
             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
+                       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
                 "#,
@@ -642,30 +770,64 @@ impl SqliteTranslationTaskRepository {
                 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) =
-                    if immutable_unchanged {
-                        (
-                            worker_status,
-                            failure_reason,
-                            attempt_count,
-                            created,
-                            completed,
-                            provider_run_id,
-                        )
-                    } else {
-                        report.refreshed_count += 1;
-                        (
-                            initial_status.as_str().to_string(),
-                            initial_failure_reason.clone(),
-                            0_i64,
-                            now,
-                            None,
-                            None,
-                        )
-                    };
+                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;
                 }
@@ -677,7 +839,11 @@ impl SqliteTranslationTaskRepository {
                         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
+                        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
                     "#,
                 )
@@ -697,6 +863,13 @@ impl SqliteTranslationTaskRepository {
                 .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)?;
@@ -770,7 +943,9 @@ impl SqliteTranslationTaskRepository {
             r#"
             SELECT task_json, worker_status, failure_reason, attempt_count,
                    created_unix_seconds, updated_unix_seconds,
-                   completed_unix_seconds, provider_run_id
+                   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
             "#,
@@ -794,6 +969,247 @@ impl SqliteTranslationTaskRepository {
         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> {
+        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 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
+            WHERE task_id = ?1
+              AND worker_status = 'running'
+              AND lease_owner = ?7
+              AND provider_run_id = ?8
+            "#,
+        )
+        .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(&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,
@@ -818,7 +1234,10 @@ impl SqliteTranslationTaskRepository {
             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_run_id = COALESCE(?7, provider_run_id),
+                lease_owner = NULL, lease_expires_unix_seconds = NULL,
+                failure_class = NULL, failure_retryable = 0,
+                next_attempt_unix_seconds = NULL
             WHERE task_id = ?1
             "#,
         )
@@ -841,7 +1260,9 @@ impl SqliteTranslationTaskRepository {
             r#"
             SELECT task_json, worker_status, failure_reason, attempt_count,
                    created_unix_seconds, updated_unix_seconds,
-                   completed_unix_seconds, provider_run_id
+                   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
             "#,
@@ -864,6 +1285,13 @@ type ExistingTaskRow = (
     i64,
     Option<i64>,
     Option<String>,
+    String,
+    Option<String>,
+    Option<String>,
+    Option<i64>,
+    Option<String>,
+    i64,
+    Option<i64>,
 );
 
 type TranslationTaskRow = (
@@ -875,6 +1303,13 @@ type TranslationTaskRow = (
     i64,
     Option<i64>,
     Option<String>,
+    String,
+    Option<String>,
+    Option<String>,
+    Option<i64>,
+    Option<String>,
+    i64,
+    Option<i64>,
 );
 
 impl PersistedTranslationTask {
@@ -889,6 +1324,13 @@ impl PersistedTranslationTask {
             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,
         }
     }
 
@@ -910,6 +1352,13 @@ impl PersistedTranslationTask {
                 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,
             },
         )
     }
@@ -924,12 +1373,21 @@ impl PersistedTranslationTask {
             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,
@@ -951,6 +1409,25 @@ impl PersistedTranslationTask {
                 })
                 .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()?,
         })
     }
 }
@@ -1219,6 +1696,140 @@ mod tests {
         assert_eq!(retrievable[0].attempt_count, 1);
     }
 
+    #[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(),
+            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![
@@ -1254,6 +1865,13 @@ mod tests {
                 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(),
@@ -1264,6 +1882,13 @@ mod tests {
                 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,
             },
         ];
 
@@ -1286,6 +1911,20 @@ mod tests {
         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();