feat(daemon): 任务执行器 + 异步 resource.sync/verify（阶段 1b）

在 daemon 里加入任务模型，让 resource.sync/verify 经 RPC 触发即返回 task_id、
可轮询进度与最终结果：

- 专用任务 worker 线程 + FIFO 队列（mpsc）；TaskRegistry 句柄包住内存态注册表
  （HashMap，有界保留 64 条已结束任务，运行中/排队中不裁剪），方法访问便于将来
  换 Redis 等持久化后端。
- 进程内 sync 锁：watch 循环与 worker 在跑同步前都获取它，互相等待而非撞
  .official-sync.lock 文件锁失败；只在实际跑同步的那段持锁，空闲睡眠不持锁。
- resource.sync/resource.verify 入队返回 { task_id, kind }（envelope status=accepted）；
  worker 复用 run_with_progress_and_cancellation（sync 用 daemon config + force；
  verify 用 dry-run+plan+audit），进度写入任务记录，should_cancel 接 daemon 停止标志
  （停机中止在途任务）。task.status/task.list 从注册表读真实数据。

验证：新增任务分发/入队/注册表单测（62 bin 测试全过）；真机端到端确认经裸 socket
触发 resource.sync 返回 task_id、轮询 task.status 走到最终态（fake curl 失败→
failed 且记录 error）、task.list 可见、resource.verify 亦入队。满足 issue #1 验收
「至少一个同步任务可经 RPC 触发并经 task.status 查询进度和最终结果」。

任务仅存内存（随 daemon 生死）；task.cancel/logs、错误码细分映射留待后续。

对应 issue #1（任务模型 + 最小方法集打通）。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

infrastructure/src/bin/bat_official_sync.rs

@@ -9,6 +9,7 @@ use bat_infrastructure::{
     OfficialVerificationSummary, OfficialVersionRecord, OfficialVersionState, PRIVATE_FILE_MODE,
 };
 use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
 use std::env;
 use std::fs::{self, OpenOptions};
 use std::io::{BufRead, BufReader, Write};
@@ -18,7 +19,7 @@ use std::os::unix::fs::OpenOptionsExt;
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::{Path, PathBuf};
 use std::process::{Command, Stdio};
-use std::sync::{Arc, Condvar, Mutex};
+use std::sync::{mpsc, Arc, Condvar, Mutex};
 use std::thread;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
@@ -267,13 +268,30 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> {
     } else {
         None
     };
-    let _rpc_server = if let Some(control) = daemon_control.as_ref() {
+    // 进程内同步锁：watch 循环与任务 worker 在跑同步前都获取它，互相等待而非撞文件锁失败。
-        Some(start_daemon_rpc_server(
+    let sync_lock = Arc::new(Mutex::new(()));
-            &daemon_state_dir,
+    let (_task_worker, _task_context, _rpc_server) = if let Some(control) = daemon_control.as_ref()
-            Arc::clone(control),
+    {
-        )?)
+        let registry = TaskRegistry::new();
+        let (task_tx, task_rx) = mpsc::channel::<TaskJob>();
+        let worker = {
+            let registry = registry.clone();
+            let sync_lock = Arc::clone(&sync_lock);
+            let control = Arc::clone(control);
+            thread::Builder::new()
+                .name("bat-daemon-task-worker".to_string())
+                .spawn(move || run_task_worker(task_rx, registry, sync_lock, control))?
+        };
+        let context = DaemonTaskContext {
+            registry,
+            queue: task_tx,
+            base_config: options.config.clone(),
+        };
+        let server =
+            start_daemon_rpc_server(&daemon_state_dir, Arc::clone(control), context.clone())?;
+        (Some(worker), Some(context), Some(server))
     } else {
-        None
+        (None, None, None)
     };
     let mut next_forced_refresh_at = next_forced_refresh_at_or_after(SystemTime::now());
     let mut pending_scheduled_force = false;
@@ -339,7 +357,12 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> {
                 next_forced_refresh_at,
             },
         );
-        match service.run_with_progress_and_cancellation(
+        // 只在实际执行同步的这段持有 sync_lock，与任务 worker 互斥；空闲睡眠时不持锁。
+        let run_result = {
+            let _sync_guard = sync_lock
+                .lock()
+                .unwrap_or_else(|poison| poison.into_inner());
+            service.run_with_progress_and_cancellation(
                 &iteration_config,
                 |event| {
                     record_daemon_progress(
@@ -351,7 +374,9 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> {
                     logger.log(event);
                 },
                 || daemon_control_stop_requested(daemon_control.as_ref()),
-        ) {
+            )
+        };
+        match run_result {
             Ok(report) => {
                 if pending_scheduled_force {
                     pending_scheduled_force = false;
@@ -610,9 +635,238 @@ const RPC_METHOD_RELOAD: &str = "daemon.reload";
 const RPC_METHOD_REFRESH: &str = "daemon.refresh";
 const RPC_METHOD_LOGS: &str = "daemon.logs";
 const RPC_METHOD_RESOURCE_STATE: &str = "resource.state";
+const RPC_METHOD_RESOURCE_SYNC: &str = "resource.sync";
+const RPC_METHOD_RESOURCE_VERIFY: &str = "resource.verify";
 const RPC_METHOD_TASK_STATUS: &str = "task.status";
 const RPC_METHOD_TASK_LIST: &str = "task.list";
 
+/// 保留的已完成任务上限（内存态，超出后裁剪最旧的已结束任务）。
+const MAX_RETAINED_TASKS: usize = 64;
+
+/// 任务类型：目前覆盖官方同步与校验。
+#[derive(Debug, Clone, Copy)]
+enum TaskKind {
+    Sync,
+    Verify,
+}
+
+impl TaskKind {
+    fn method(self) -> &'static str {
+        match self {
+            Self::Sync => RPC_METHOD_RESOURCE_SYNC,
+            Self::Verify => RPC_METHOD_RESOURCE_VERIFY,
+        }
+    }
+
+    /// 由 daemon 基准配置派生该任务的实际同步配置。
+    fn build_config(self, base: &OfficialUpdateConfig, force: bool) -> OfficialUpdateConfig {
+        let mut config = base.clone();
+        match self {
+            Self::Sync => {
+                config.dry_run = false;
+                config.force = config.force || force;
+            }
+            Self::Verify => {
+                config.dry_run = true;
+                config.plan = true;
+                config.audit_local = true;
+                config.repair = false;
+                config.force = false;
+            }
+        }
+        config
+    }
+}
+
+/// 单个任务的可轮询记录。
+#[derive(Debug, Clone, Serialize)]
+struct TaskRecord {
+    id: String,
+    kind: &'static str,
+    /// `queued` | `running` | `succeeded` | `failed`。
+    status: &'static str,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    stage: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    message: Option<String>,
+    created_at: u64,
+    updated_at: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    started_at: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    finished_at: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    error: Option<ApiError>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    result: Option<serde_json::Value>,
+}
+
+struct TaskStore {
+    tasks: HashMap<String, TaskRecord>,
+    order: Vec<String>,
+    seq: u64,
+}
+
+/// 任务注册表句柄：包住内存存储，供 RPC handler 与 worker 共享。
+///
+/// 通过方法访问（而非直接摸内部 map），便于将来换成 Redis 等持久化后端。
+#[derive(Clone)]
+struct TaskRegistry {
+    inner: Arc<Mutex<TaskStore>>,
+}
+
+impl TaskRegistry {
+    fn new() -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(TaskStore {
+                tasks: HashMap::new(),
+                order: Vec::new(),
+                seq: 0,
+            })),
+        }
+    }
+
+    fn lock(&self) -> std::sync::MutexGuard<'_, TaskStore> {
+        self.inner
+            .lock()
+            .unwrap_or_else(|poison| poison.into_inner())
+    }
+
+    /// 创建 queued 任务并返回 task_id。
+    fn create(&self, kind: TaskKind) -> String {
+        let now = unix_seconds_now();
+        let mut store = self.lock();
+        store.seq += 1;
+        let id = format!("task-{}-{}", std::process::id(), store.seq);
+        let record = TaskRecord {
+            id: id.clone(),
+            kind: kind.method(),
+            status: "queued",
+            stage: None,
+            message: None,
+            created_at: now,
+            updated_at: now,
+            started_at: None,
+            finished_at: None,
+            error: None,
+            result: None,
+        };
+        store.tasks.insert(id.clone(), record);
+        store.order.push(id.clone());
+        store.prune();
+        id
+    }
+
+    fn update<F: FnOnce(&mut TaskRecord)>(&self, id: &str, update: F) {
+        let mut store = self.lock();
+        if let Some(record) = store.tasks.get_mut(id) {
+            update(record);
+            record.updated_at = unix_seconds_now();
+        }
+    }
+
+    fn get(&self, id: &str) -> Option<TaskRecord> {
+        self.lock().tasks.get(id).cloned()
+    }
+
+    /// 返回全部任务，最新创建的在前。
+    fn list(&self) -> Vec<TaskRecord> {
+        let store = self.lock();
+        store
+            .order
+            .iter()
+            .rev()
+            .filter_map(|id| store.tasks.get(id).cloned())
+            .collect()
+    }
+}
+
+impl TaskStore {
+    /// 裁剪最旧的已结束任务，把内存占用控制在上限内；运行中/排队中的任务不裁剪。
+    fn prune(&mut self) {
+        while self.order.len() > MAX_RETAINED_TASKS {
+            let Some(position) = self.order.iter().position(|id| {
+                self.tasks
+                    .get(id)
+                    .map(|record| matches!(record.status, "succeeded" | "failed"))
+                    .unwrap_or(true)
+            }) else {
+                break;
+            };
+            let id = self.order.remove(position);
+            self.tasks.remove(&id);
+        }
+    }
+}
+
+/// 提交给任务 worker 的作业（配置已按任务类型派生完毕）。
+struct TaskJob {
+    id: String,
+    config: OfficialUpdateConfig,
+}
+
+/// daemon 任务上下文：RPC handler 借它创建任务、入队和读取。
+#[derive(Clone)]
+struct DaemonTaskContext {
+    registry: TaskRegistry,
+    queue: mpsc::Sender<TaskJob>,
+    base_config: OfficialUpdateConfig,
+}
+
+/// 任务 worker：单线程 FIFO 消费任务队列，串行执行官方同步/校验。
+///
+/// 每个任务执行前获取进程内 `sync_lock`，与 watch 循环互斥（等待而非撞文件锁失败）；
+/// 进度写入任务记录；`should_cancel` 接 daemon 停止标志，停机时中止在途任务。
+fn run_task_worker(
+    receiver: mpsc::Receiver<TaskJob>,
+    registry: TaskRegistry,
+    sync_lock: Arc<Mutex<()>>,
+    control: DaemonControl,
+) {
+    let service = OfficialUpdateService::new();
+    for job in receiver {
+        registry.update(&job.id, |record| {
+            record.status = "running";
+            record.started_at = Some(unix_seconds_now());
+        });
+
+        let run_result = {
+            let _sync_guard = sync_lock
+                .lock()
+                .unwrap_or_else(|poison| poison.into_inner());
+            let progress_registry = registry.clone();
+            let progress_id = job.id.clone();
+            service.run_with_progress_and_cancellation(
+                &job.config,
+                |event| {
+                    progress_registry.update(&progress_id, |record| {
+                        record.stage = Some(event.stage.to_string());
+                        record.message = Some(event.message.clone());
+                    });
+                },
+                || daemon_control_stop_requested(Some(&control)),
+            )
+        };
+
+        match run_result {
+            Ok(report) => registry.update(&job.id, |record| {
+                record.status = "succeeded";
+                record.finished_at = Some(unix_seconds_now());
+                record.result = serde_json::to_value(&report).ok();
+            }),
+            Err(error) => registry.update(&job.id, |record| {
+                record.status = "failed";
+                record.finished_at = Some(unix_seconds_now());
+                record.error = Some(ApiError::new(
+                    ErrorCode::INTERNAL,
+                    "task.executor",
+                    error.to_string(),
+                ));
+            }),
+        }
+    }
+}
+
 /// 把 `bat.*` 兼容别名解析为规范的 `daemon.*` 方法名；其余原样返回。
 fn canonical_rpc_method(method: &str) -> &str {
     match method {
@@ -627,10 +881,8 @@ fn canonical_rpc_method(method: &str) -> &str {
 
 /// 判断方法是否属于已规划但尚未实现的命名空间/动作（返回 not_implemented 而非 unknown）。
 fn is_pending_rpc_method(method: &str) -> bool {
-    matches!(
+    matches!(method, "task.create" | "task.cancel" | "task.logs")
-        method,
+        || method.starts_with("catalog.")
-        "resource.sync" | "resource.verify" | "task.create" | "task.cancel" | "task.logs"
-    ) || method.starts_with("catalog.")
         || method.starts_with("patch.")
         || method.starts_with("unityfs.")
 }
@@ -1058,6 +1310,7 @@ fn take_daemon_wake(state: &mut DaemonControlState) -> Option<DaemonWake> {
 fn start_daemon_rpc_server(
     state_dir: &Path,
     control: DaemonControl,
+    tasks: DaemonTaskContext,
 ) -> anyhow::Result<DaemonRpcServer> {
     validate_runtime_state_dir(state_dir).map_err(anyhow::Error::msg)?;
     fs::create_dir_all(state_dir)?;
@@ -1087,9 +1340,12 @@ fn start_daemon_rpc_server(
                     Ok(stream) => {
                         let state_dir = server_state_dir.clone();
                         let control = Arc::clone(&control);
+                        let tasks = tasks.clone();
                         let _ = thread::Builder::new()
                             .name("bat-daemon-rpc-client".to_string())
-                            .spawn(move || handle_daemon_rpc_client(stream, state_dir, control));
+                            .spawn(move || {
+                                handle_daemon_rpc_client(stream, state_dir, control, tasks)
+                            });
                     }
                     Err(error) => {
                         eprintln!("[daemon] RPC socket accept 失败：{error}");
@@ -1106,12 +1362,18 @@ fn start_daemon_rpc_server(
 fn start_daemon_rpc_server(
     _state_dir: &Path,
     _control: DaemonControl,
+    _tasks: DaemonTaskContext,
 ) -> anyhow::Result<DaemonRpcServer> {
     Err(anyhow::anyhow!("daemon RPC 目前只支持 Unix/Linux 平台"))
 }
 
 #[cfg(unix)]
-fn handle_daemon_rpc_client(mut stream: UnixStream, state_dir: PathBuf, control: DaemonControl) {
+fn handle_daemon_rpc_client(
+    mut stream: UnixStream,
+    state_dir: PathBuf,
+    control: DaemonControl,
+    tasks: DaemonTaskContext,
+) {
     let Ok(reader_stream) = stream.try_clone() else {
         return;
     };
@@ -1137,7 +1399,7 @@ fn handle_daemon_rpc_client(mut stream: UnixStream, state_dir: PathBuf, control:
         let response = match serde_json::from_str::<JsonRpcRequest>(&line) {
             Ok(request) => {
                 notify_stop_after_response = request.method == RPC_METHOD_STOP;
-                handle_daemon_rpc_request(request, &state_dir, &control)
+                handle_daemon_rpc_request(request, &state_dir, &control, &tasks)
             }
             Err(error) => json_rpc_error(None, -32700, format!("JSON-RPC 请求解析失败：{error}")),
         };
@@ -1158,10 +1420,11 @@ fn handle_daemon_rpc_request(
     request: JsonRpcRequest,
     state_dir: &Path,
     control: &DaemonControl,
+    tasks: &DaemonTaskContext,
 ) -> JsonRpcResponse {
     let id = request.id.clone();
     let request_id = next_request_id();
-    let envelope = dispatch_rpc_method(&request, state_dir, control, request_id);
+    let envelope = dispatch_rpc_method(&request, state_dir, control, tasks, request_id);
     match serde_json::to_value(&envelope) {
         Ok(value) => json_rpc_result(id, value),
         Err(error) => json_rpc_error(id, -32603, error.to_string()),
@@ -1173,6 +1436,7 @@ fn dispatch_rpc_method(
     request: &JsonRpcRequest,
     state_dir: &Path,
     control: &DaemonControl,
+    tasks: &DaemonTaskContext,
     request_id: String,
 ) -> RpcEnvelope {
     match canonical_rpc_method(&request.method) {
@@ -1244,6 +1508,13 @@ fn dispatch_rpc_method(
             "resource.state",
             build_resource_state_report(state_dir),
         ),
+        RPC_METHOD_RESOURCE_SYNC => {
+            let force = rpc_bool_param(request.params.as_ref(), "force").unwrap_or(false);
+            enqueue_task_envelope(tasks, TaskKind::Sync, force, request_id)
+        }
+        RPC_METHOD_RESOURCE_VERIFY => {
+            enqueue_task_envelope(tasks, TaskKind::Verify, false, request_id)
+        }
         RPC_METHOD_TASK_STATUS => {
             let task_id = request
                 .params
@@ -1251,19 +1522,28 @@ fn dispatch_rpc_method(
                 .and_then(|params| params.get("task_id"))
                 .and_then(serde_json::Value::as_str)
                 .unwrap_or_default();
-            // 任务执行器尚未落地（阶段 1b）；任务库为空，任何 task_id 都不存在。
+            match tasks.registry.get(task_id) {
-            rpc_envelope_error(
+                Some(record) => rpc_envelope_from_result(
+                    request_id,
+                    "task.status",
+                    serde_json::to_value(record).map_err(anyhow::Error::from),
+                ),
+                None => rpc_envelope_error(
                     request_id,
                     ApiError::new(
                         ErrorCode::TASK_NOT_FOUND,
                         "task.status",
                         format!("任务不存在：{task_id}"),
                     ),
-            )
+                ),
             }
-        RPC_METHOD_TASK_LIST => {
-            rpc_envelope_ok(request_id, "ok", serde_json::json!({ "tasks": [] }))
         }
+        RPC_METHOD_TASK_LIST => rpc_envelope_from_result(
+            request_id,
+            "task.list",
+            serde_json::to_value(serde_json::json!({ "tasks": tasks.registry.list() }))
+                .map_err(anyhow::Error::from),
+        ),
         pending if is_pending_rpc_method(pending) => rpc_envelope_error(
             request_id,
             ApiError::new(
@@ -1283,6 +1563,42 @@ fn dispatch_rpc_method(
     }
 }
 
+/// 创建任务、入队，返回 `accepted` + task_id 的 envelope。
+fn enqueue_task_envelope(
+    tasks: &DaemonTaskContext,
+    kind: TaskKind,
+    force: bool,
+    request_id: String,
+) -> RpcEnvelope {
+    let config = kind.build_config(&tasks.base_config, force);
+    let task_id = tasks.registry.create(kind);
+    let job = TaskJob {
+        id: task_id.clone(),
+        config,
+    };
+    if tasks.queue.send(job).is_err() {
+        // worker 已退出：把该任务标记为失败并返回错误。
+        tasks.registry.update(&task_id, |record| {
+            record.status = "failed";
+            record.finished_at = Some(unix_seconds_now());
+            record.error = Some(ApiError::new(
+                ErrorCode::INTERNAL,
+                "task.enqueue",
+                "任务执行器不可用",
+            ));
+        });
+        return rpc_envelope_error(
+            request_id,
+            ApiError::new(ErrorCode::INTERNAL, "task.enqueue", "任务执行器不可用"),
+        );
+    }
+    rpc_envelope_ok(
+        request_id,
+        "accepted",
+        serde_json::json!({ "task_id": task_id, "kind": kind.method() }),
+    )
+}
+
 fn rpc_ack_value(
     command: &'static str,
     message: &'static str,
@@ -4901,13 +5217,23 @@ mod tests {
 
     #[test]
     fn is_pending_rpc_method_covers_planned_namespaces() {
-        assert!(is_pending_rpc_method("resource.sync"));
-        assert!(is_pending_rpc_method("resource.verify"));
         assert!(is_pending_rpc_method("catalog.status"));
         assert!(is_pending_rpc_method("patch.apply"));
         assert!(is_pending_rpc_method("unityfs.inspect"));
+        assert!(is_pending_rpc_method("task.cancel"));
+        // sync/verify 已由任务执行器实现，不再是 pending。
+        assert!(!is_pending_rpc_method("resource.sync"));
+        assert!(!is_pending_rpc_method("resource.verify"));
         assert!(!is_pending_rpc_method("daemon.status"));
-        assert!(!is_pending_rpc_method("resource.state"));
+    }
+
+    fn test_task_context() -> DaemonTaskContext {
+        let (queue, _rx) = mpsc::channel::<TaskJob>();
+        DaemonTaskContext {
+            registry: TaskRegistry::new(),
+            queue,
+            base_config: OfficialUpdateConfig::default(),
+        }
     }
 
     #[test]
@@ -4918,6 +5244,7 @@ mod tests {
             &rpc_request("nope.nope", None),
             temp.path(),
             &control,
+            &test_task_context(),
             "req-test-1".to_string(),
         );
         let value = serde_json::to_value(&envelope).unwrap();
@@ -4931,9 +5258,10 @@ mod tests {
         let temp = tempfile::TempDir::new().unwrap();
         let control = new_daemon_control();
         let envelope = dispatch_rpc_method(
-            &rpc_request("resource.sync", None),
+            &rpc_request("catalog.status", None),
             temp.path(),
             &control,
+            &test_task_context(),
             "req-test-2".to_string(),
         );
         let value = serde_json::to_value(&envelope).unwrap();
@@ -4949,6 +5277,7 @@ mod tests {
             &rpc_request("task.list", None),
             temp.path(),
             &control,
+            &test_task_context(),
             "req-test-3".to_string(),
         );
         let value = serde_json::to_value(&envelope).unwrap();
@@ -4965,6 +5294,7 @@ mod tests {
             &rpc_request("task.status", Some(serde_json::json!({ "task_id": "abc" }))),
             temp.path(),
             &control,
+            &test_task_context(),
             "req-test-4".to_string(),
         );
         let value = serde_json::to_value(&envelope).unwrap();
@@ -4972,6 +5302,77 @@ mod tests {
         assert_eq!(value["error"]["code"], "BAT-ERR-700004");
     }
 
+    #[test]
+    fn dispatch_resource_sync_enqueues_task() {
+        let temp = tempfile::TempDir::new().unwrap();
+        let control = new_daemon_control();
+        // 保留 rx 让 send 成功（不启动 worker，任务停留在 queued）。
+        let (queue, rx) = mpsc::channel::<TaskJob>();
+        let context = DaemonTaskContext {
+            registry: TaskRegistry::new(),
+            queue,
+            base_config: OfficialUpdateConfig::default(),
+        };
+
+        let envelope = dispatch_rpc_method(
+            &rpc_request("resource.sync", Some(serde_json::json!({ "force": true }))),
+            temp.path(),
+            &control,
+            &context,
+            "req-test-5".to_string(),
+        );
+        let value = serde_json::to_value(&envelope).unwrap();
+        assert_eq!(value["ok"], true);
+        assert_eq!(value["status"], "accepted");
+        let task_id = value["data"]["task_id"].as_str().unwrap().to_string();
+        assert_eq!(value["data"]["kind"], "resource.sync");
+
+        // 注册表里应有一条 queued 任务。
+        let record = context.registry.get(&task_id).unwrap();
+        assert_eq!(record.status, "queued");
+        assert_eq!(record.kind, "resource.sync");
+
+        // 作业已入队，force 已透传到派生配置。
+        let job = rx.try_recv().unwrap();
+        assert_eq!(job.id, task_id);
+        assert!(job.config.force);
+        assert!(!job.config.dry_run);
+
+        // task.status 能查到这条任务。
+        let status_envelope = dispatch_rpc_method(
+            &rpc_request(
+                "task.status",
+                Some(serde_json::json!({ "task_id": task_id })),
+            ),
+            temp.path(),
+            &control,
+            &context,
+            "req-test-6".to_string(),
+        );
+        let status_value = serde_json::to_value(&status_envelope).unwrap();
+        assert_eq!(status_value["ok"], true);
+        assert_eq!(status_value["data"]["status"], "queued");
+    }
+
+    #[test]
+    fn task_registry_create_update_and_prune() {
+        let registry = TaskRegistry::new();
+        let id = registry.create(TaskKind::Verify);
+        assert_eq!(registry.get(&id).unwrap().status, "queued");
+        registry.update(&id, |record| record.status = "running");
+        assert_eq!(registry.get(&id).unwrap().status, "running");
+        assert_eq!(registry.list().len(), 1);
+
+        // 超过上限时裁剪最旧的已结束任务；未结束的保留。
+        for _ in 0..MAX_RETAINED_TASKS {
+            let finished = registry.create(TaskKind::Sync);
+            registry.update(&finished, |record| record.status = "succeeded");
+        }
+        assert!(registry.list().len() <= MAX_RETAINED_TASKS);
+        // 运行中的任务不被裁剪。
+        assert_eq!(registry.get(&id).unwrap().status, "running");
+    }
+
     #[test]
     fn parses_explicit_watch_error_retry_interval() {
         let options =