feat(mobile): upload engine with preparation, batching states and retries

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

apps/mobile/lib/features/uploads/models.dart

@@ -81,8 +81,9 @@ class UploadCredential {
 /// `initUpload`，POST /uploads）：幂等初始化——相同 clientTaskId 且参数
 /// 一致时返回原会话，参数不一致 409（UPLOAD_SESSION_CONFLICT）。
 ///
-/// 注：契约增补（P3 计划 Task 3）将顺带加入可选 sourceDeviceId
-/// （A19 设备来源语义）；当前以契约现状为准，不含该字段。
+/// sourceDeviceId 为 P3 Task 3 契约增补（已落地）：客户端首次生成并
+/// 持久的设备标识（A19 设备来源语义，见 core/device_id.dart）；可选，
+/// 同一 clientTaskId 幂等重放时服务端保留首次落库的值。
 class UploadInitRequest {
   const UploadInitRequest({
     required this.albumId,
@@ -91,6 +92,7 @@ class UploadInitRequest {
     required this.bytes,
     required this.contentType,
     this.categoryId,
+    this.sourceDeviceId,
   });
 
   /// 目标相册 ID。
@@ -111,6 +113,9 @@ class UploadInitRequest {
   /// MIME 类型（白名单 image/jpeg | image/png | image/heic）。
   final String contentType;
 
+  /// 发起设备标识（可选，minLength 8 / maxLength 64；A19）。
+  final String? sourceDeviceId;
+
   Map<String, dynamic> toJson() => {
         'albumId': albumId,
         if (categoryId != null) 'categoryId': categoryId,
@@ -118,6 +123,7 @@ class UploadInitRequest {
         'filename': filename,
         'bytes': bytes,
         'contentType': contentType,
+        if (sourceDeviceId != null) 'sourceDeviceId': sourceDeviceId,
       };
 }
 
@@ -243,6 +249,7 @@ class UploadSessionRecord {
     required this.contentType,
     required this.createdAt,
     required this.photo,
+    this.sourceDeviceId,
   });
 
   factory UploadSessionRecord.fromJson(Map<String, dynamic> json) =>
@@ -257,6 +264,7 @@ class UploadSessionRecord {
         photo: UploadSessionPhoto.fromJson(
           json['photo'] as Map<String, dynamic>,
         ),
+        sourceDeviceId: json['sourceDeviceId'] as String?,
       );
 
   /// 上传会话 ID。
@@ -280,6 +288,9 @@ class UploadSessionRecord {
   /// 会话创建时间。
   final DateTime createdAt;
 
+  /// 发起设备标识（init 时未携带为 null；P3 Task 3 契约增补，A19）。
+  final String? sourceDeviceId;
+
   /// 会话关联照片的当前状态。
   final UploadSessionPhoto photo;
 
@@ -291,6 +302,7 @@ class UploadSessionRecord {
         'expectedBytes': expectedBytes,
         'contentType': contentType,
         'createdAt': createdAt.toUtc().toIso8601String(),
+        if (sourceDeviceId != null) 'sourceDeviceId': sourceDeviceId,
         'photo': photo.toJson(),
       };
 }

apps/mobile/lib/features/uploads/preparer.dart

@@ -0,0 +1,306 @@
+/// 上传准备阶段（P3 计划 Task 6）。
+///
+/// 选图（image_picker）或连拍（camera）得到的源文件 → 复制到受管目录
+/// （app 支持目录/uploads/{accountId}/{albumId}/{taskId}{ext}）→ stat 字节
+/// → MIME 嗅探（扩展名映射 + jpg/png 魔数校验）→ 尺寸上限本地预检 →
+/// 入队 queued。任一步失败 → 独立状态 preparationFailed（可重试，见
+/// [UploadPreparer.retry]），lastError 携带客户端伪码（[PrepErrorCodes]）。
+///
+/// 设计取舍（对齐计划 Task 6 评审口径）：
+/// - **串行执行**：文件复制是最容易造成 IO 风暴的环节（连拍场景一次
+///   落盘多张），队列化逐张处理；传输并发由 UploadEngine 另行控制；
+/// - **磁盘空间不显式预检**：引入 disk_space 插件成本（新依赖 + 平台
+///   误差）高于收益——复制失败本身即失败信号（ENOSPC 走 copy 失败分支
+///   → preparationFailed，用户可见可重试）。此为有意裁剪，登记为计划
+///   偏差；
+/// - **本地尺寸上限**：客户端无法得知服务端 UPLOAD_MAX 环境变量，取
+///   保守常量 50MB 预检（超限直接 preparationFailed，不浪费复制与传输）；
+///   界于本地上限与服务端真实上限之间的文件由服务端 400 兜底。
+library;
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:math';
+
+import 'package:path/path.dart' as p;
+import 'package:path_provider/path_provider.dart';
+
+import '../../core/device_id.dart';
+import 'task_state.dart';
+import 'task_store.dart';
+
+/// 准备阶段失败伪码（客户端本地，契约外；lastError 展示用，
+/// UI 层映射本地化文案）。
+abstract final class PrepErrorCodes {
+  /// 源文件不可读（已被清理/权限/不存在）。
+  static const String sourceUnreadable = 'PREP_SOURCE_UNREADABLE';
+
+  /// MIME 不在白名单（jpg/png/heic 之外）。
+  static const String unsupportedType = 'PREP_UNSUPPORTED_TYPE';
+
+  /// 超过本地尺寸上限（50MB 预检）。
+  static const String tooLarge = 'PREP_TOO_LARGE';
+
+  /// 复制到受管目录失败（含磁盘满/复制期间源被改动）。
+  static const String copyFailed = 'PREP_COPY_FAILED';
+}
+
+/// 上传准备器：入队唯一入口（队列 UI/连拍页调用；引擎消费产出）。
+///
+/// 所有操作串行（连拍入队逐张复制，避免 IO 风暴，见库级注释）。
+class UploadPreparer {
+  UploadPreparer(
+    this._store, {
+    Future<Directory> Function()? baseDirProvider,
+    Random? random,
+    this.maxBytes = kMaxUploadBytes,
+    this.onEnqueued,
+  })  : _baseDirProvider = baseDirProvider ?? getApplicationSupportDirectory,
+        _random = random ?? Random.secure();
+
+  final TaskStore _store;
+  final Future<Directory> Function() _baseDirProvider;
+  final Random _random;
+
+  /// 本地尺寸预检上限（默认 [kMaxUploadBytes]；测试可注入小值）。
+  final int maxBytes;
+
+  /// 本地尺寸预检上限（50MB；服务端 UPLOAD_MAX 由 400 兜底，见库级注释）。
+  static const int kMaxUploadBytes = 50 * 1024 * 1024;
+
+  /// 入队成功回调（唤醒传输引擎立即出队）。
+  final void Function()? onEnqueued;
+
+  /// 串行队列尾部。
+  Future<void> _tail = Future<void>.value();
+
+  /// 准备一张图片并入队。
+  ///
+  /// [sourcePath] 为 image_picker/camera 返回的源文件路径（App 沙箱内
+  /// 可直接 File 读取）；[filename] 缺省取源文件 basename。
+  ///
+  /// 返回入队后的任务（state=queued）；失败返回 state=preparationFailed
+  /// 的任务（不抛异常——准备失败是业务状态而非调用错误）。
+  Future<UploadTask> prepare({
+    required String accountId,
+    required String albumId,
+    String? categoryId,
+    required String sourcePath,
+    String? filename,
+  }) {
+    return _serialize(() => _prepare(
+          accountId: accountId,
+          albumId: albumId,
+          categoryId: categoryId,
+          sourcePath: sourcePath,
+          filename: filename,
+        ));
+  }
+
+  /// 重试准备失败的任务（preparationFailed → preparing → 复制 → queued）。
+  ///
+  /// 源文件原始路径不落库（任务表仅存受管副本路径），调用方（队列 UI）
+  /// 需重新提供 [sourcePath]（重新选图，或相机缓存仍存时直传）。
+  Future<UploadTask> retry({
+    required String accountId,
+    required String taskId,
+    required String sourcePath,
+  }) {
+    return _serialize(() => _retry(
+          accountId: accountId,
+          taskId: taskId,
+          sourcePath: sourcePath,
+        ));
+  }
+
+  /// 串行化：所有准备操作首尾相接（错误吞掉以不断链，错误经返回值上抛）。
+  Future<T> _serialize<T>(Future<T> Function() op) {
+    final result = _tail.then((_) => op());
+    _tail = result.then<void>((_) {}, onError: (Object _) {});
+    return result;
+  }
+
+  Future<UploadTask> _prepare({
+    required String accountId,
+    required String albumId,
+    required String? categoryId,
+    required String sourcePath,
+    required String? filename,
+  }) async {
+    final id = _newId();
+    final clientTaskId = _newId();
+    final ext = _normalizeExt(sourcePath);
+    final contentType = await _sniffContentType(sourcePath, ext);
+    final bytes = await _statBytes(sourcePath);
+
+    // 快速失败预检：类型不在白名单 / 源不可读（bytes=0）/ 超本地上限。
+    final failCode = _preCheck(contentType: contentType, bytes: bytes);
+    final dir = await _managedDir(accountId, albumId);
+    final targetPath = p.join(dir.path, '$id$ext');
+
+    final task = await _store.create(
+      UploadTaskDraft(
+        id: id,
+        accountId: accountId,
+        albumId: albumId,
+        categoryId: categoryId,
+        clientTaskId: clientTaskId,
+        sourceFilePath: targetPath,
+        filename: filename ?? p.basename(sourcePath),
+        // 预检失败时占位（该行永不进入传输；成功路径 bytes 来自源 stat）。
+        bytes: bytes > 0 ? bytes : 1,
+        contentType: contentType ?? 'application/octet-stream',
+      ),
+    );
+
+    if (failCode != null) {
+      return _store.updateState(
+        id,
+        UploadTaskState.preparationFailed,
+        patch: UploadTaskPatch(lastError: failCode),
+      );
+    }
+
+    return _copyAndQueue(task, sourcePath: sourcePath);
+  }
+
+  Future<UploadTask> _retry({
+    required String accountId,
+    required String taskId,
+    required String sourcePath,
+  }) async {
+    final task = await _store.get(accountId, taskId);
+    if (task == null) {
+      throw StateError('上传任务不存在：$taskId');
+    }
+    if (task.state != UploadTaskState.preparationFailed) {
+      throw StateError('仅 preparationFailed 可重试准备（当前：${task.state}）');
+    }
+    final preparing = await _store.updateState(
+      taskId,
+      UploadTaskState.preparing,
+      patch: const UploadTaskPatch(clearLastError: true),
+    );
+    return _copyAndQueue(preparing, sourcePath: sourcePath);
+  }
+
+  /// 复制到受管路径并入队；复制失败 → preparationFailed（返回，不抛）。
+  Future<UploadTask> _copyAndQueue(
+    UploadTask task, {
+    required String sourcePath,
+  }) async {
+    final source = File(sourcePath);
+    final target = File(task.sourceFilePath);
+    try {
+      final expected = await source.length();
+      if (expected <= 0) {
+        throw StateError('empty source');
+      }
+      await target.parent.create(recursive: true);
+      await source.copy(task.sourceFilePath);
+      final copied = await target.length();
+      if (copied != expected) {
+        // 源在复制期间被改动（相机流水写入中）——按复制失败处理，
+        // 用户可重试。
+        if (await target.exists()) {
+          await target.delete();
+        }
+        throw StateError('copy size mismatch');
+      }
+      // 声明值与副本对齐（重试换源场景下行内 bytes 可能是占位值；
+      // TaskStore.updateMeta 旁路写入，不迁移状态）。
+      if (copied != task.bytes) {
+        await _store.updateMeta(task.id, bytes: copied);
+      }
+    } on Object {
+      return _store.updateState(
+        task.id,
+        UploadTaskState.preparationFailed,
+        patch: const UploadTaskPatch(lastError: PrepErrorCodes.copyFailed),
+      );
+    }
+    final queued = await _store.updateState(
+      task.id,
+      UploadTaskState.queued,
+      patch: const UploadTaskPatch(clearLastError: true),
+    );
+    onEnqueued?.call();
+    return queued;
+  }
+
+  /// 快速预检：返回失败伪码（null=通过）。
+  String? _preCheck({required String? contentType, required int bytes}) {
+    if (bytes <= 0) return PrepErrorCodes.sourceUnreadable;
+    if (contentType == null) return PrepErrorCodes.unsupportedType;
+    if (bytes > maxBytes) return PrepErrorCodes.tooLarge;
+    return null;
+  }
+
+  /// MIME 嗅探：魔数优先（jpg/png），回退扩展名映射（含无魔数的 heic）。
+  ///
+  /// 返回 null=无法确定白名单内类型（调用方按 unsupportedType 失败）。
+  Future<String?> _sniffContentType(String sourcePath, String ext) async {
+    try {
+      final head = await File(sourcePath).openRead(0, 8).first;
+      if (head.length >= 3 &&
+          head[0] == 0xFF &&
+          head[1] == 0xD8 &&
+          head[2] == 0xFF) {
+        return 'image/jpeg';
+      }
+      if (head.length >= 4 &&
+          head[0] == 0x89 &&
+          head[1] == 0x50 &&
+          head[2] == 0x4E &&
+          head[3] == 0x47) {
+        return 'image/png';
+      }
+    } on Object {
+      // 魔数读取失败（源不存在/权限）：交由 _statBytes→sourceUnreadable。
+    }
+    return _contentTypeByExt(ext);
+  }
+
+  /// 扩展名 → 白名单 MIME（未知扩展返回 null）。
+  static String? _contentTypeByExt(String ext) {
+    switch (ext) {
+      case '.jpg':
+      case '.jpeg':
+        return 'image/jpeg';
+      case '.png':
+        return 'image/png';
+      case '.heic':
+      case '.heif':
+        return 'image/heic';
+    }
+    return null;
+  }
+
+  /// stat 源字节数；不可读返回 0（触发 sourceUnreadable 预检失败）。
+  Future<int> _statBytes(String sourcePath) async {
+    try {
+      return await File(sourcePath).length();
+    } on Object {
+      return 0;
+    }
+  }
+
+  /// 归一小写扩展名（含点；无扩展返回空串）。
+  static String _normalizeExt(String sourcePath) =>
+      p.extension(sourcePath).toLowerCase();
+
+  /// 受管目录：{base}/uploads/{accountId}/{albumId}/。
+  Future<Directory> _managedDir(String accountId, String albumId) async {
+    final base = await _baseDirProvider();
+    final dir = Directory(p.join(base.path, 'uploads', accountId, albumId));
+    await dir.create(recursive: true);
+    return dir;
+  }
+
+  /// 32 位十六进制随机 ID（任务 id 与 clientTaskId 各一次生成，
+  /// 与 core/device_id.dart 同一编码方式）。
+  String _newId() {
+    final bytes = List<int>.generate(16, (_) => _random.nextInt(256));
+    return DeviceId.hexEncode(bytes);
+  }
+}

apps/mobile/lib/features/uploads/providers.dart

@@ -0,0 +1,43 @@
+/// 上传域 Provider 装配（P3 计划 Task 6）。
+///
+/// 生产装配：TaskStore.app()（documents/upload_tasks.sqlite）+ ApiClient
+/// （Bearer 自动注入）+ DeviceId（shared_preferences）。引擎生命周期
+/// （start(accountId)/stop，随登录态切换）由 App 壳/队列 UI 驱动——
+/// Task 9 队列 UI 落地时接线；本文件保证依赖图就绪。
+library;
+
+import 'package:flutter_riverpod/flutter_riverpod.dart';
+
+import '../../core/api_client.dart';
+import '../../core/device_id.dart';
+import 'preparer.dart';
+import 'task_store.dart';
+import 'upload_engine.dart';
+
+/// 任务库（生产：app documents dir 持久库；测试 override 为 memory）。
+final taskStoreProvider = Provider<TaskStore>((ref) {
+  final store = TaskStore.app();
+  ref.onDispose(store.close);
+  return store;
+});
+
+/// 传输引擎（裸 Dio 直传 presign；并行/保留期配置见 upload_engine.dart
+/// 的 dart-define 常量）。
+final uploadEngineProvider = Provider<UploadEngine>((ref) {
+  final engine = UploadEngine(
+    ref.watch(taskStoreProvider),
+    ref.watch(apiClientProvider),
+    ref.watch(deviceIdProvider),
+  );
+  ref.onDispose(engine.dispose);
+  return engine;
+});
+
+/// 准备器（入队回调唤醒引擎立即出队）。
+final uploadPreparerProvider = Provider<UploadPreparer>((ref) {
+  final engine = ref.watch(uploadEngineProvider);
+  return UploadPreparer(
+    ref.watch(taskStoreProvider),
+    onEnqueued: engine.notifyWorkAvailable,
+  );
+});

apps/mobile/lib/features/uploads/task_store.dart

@@ -292,6 +292,10 @@ class TaskStore {
 
   /// 状态迁移（带合法迁移守护 + updatedAt 刷新），返回迁移后的任务。
   ///
+  /// patch 语义（与 [UploadTaskPatch] 文档一致）：**未提供的列保持不变**，
+  /// 显式清除走 clear* 布尔；serverUploadId/serverPhotoId 仅在提供时
+  /// 写入（write-once 回填，无清除通道）。
+  ///
   /// 抛 [StateError]：任务不存在，或 `当前 → to` 不在合法迁移表
   /// （[assertTransition]）。
   Future<UploadTask> updateState(
@@ -313,12 +317,54 @@ class TaskStore {
               : Value(patch.attempt!),
           nextRetryAt: patch.clearNextRetryAt
               ? const Value(null)
-              : Value(patch.nextRetryAt?.toUtc()),
-          serverUploadId: Value(patch.serverUploadId),
-          serverPhotoId: Value(patch.serverPhotoId),
+              : patch.nextRetryAt == null
+                  ? const Value.absent()
+                  : Value(patch.nextRetryAt!.toUtc()),
+          serverUploadId: patch.serverUploadId == null
+              ? const Value.absent()
+              : Value(patch.serverUploadId),
+          serverPhotoId: patch.serverPhotoId == null
+              ? const Value.absent()
+              : Value(patch.serverPhotoId),
           lastError: patch.clearLastError
               ? const Value(null)
-              : Value(patch.lastError),
+              : patch.lastError == null
+                  ? const Value.absent()
+                  : Value(patch.lastError),
+          updatedAt: Value(_utcAt(now)),
+        ),
+      );
+      return await _byId(id);
+    });
+  }
+
+  /// 旁路字段回填（**不迁移状态**）：引擎在 initUpload 成功后写
+  /// serverUploadId/serverPhotoId——此时状态已为 uploading，无合法自迁移，
+  /// 需绕过状态机直接回填（仅刷新 updatedAt）。bytes 供准备器重试换源后
+  /// 对齐声明值（副本 stat 与行内声明可能有历史偏差）。
+  ///
+  /// 提供的字段写入，未提供的保持不变（与 [updateState] 的 patch 语义
+  /// 一致）。
+  Future<UploadTask> updateMeta(
+    String id, {
+    String? serverUploadId,
+    String? serverPhotoId,
+    int? bytes,
+    DateTime? now,
+  }) {
+    return _db.transaction(() async {
+      await _byId(id);
+      await (_db.update(_db.uploadTasks)
+            ..where((t) => t.id.equals(id)))
+          .write(
+        UploadTasksCompanion(
+          serverUploadId: serverUploadId == null
+              ? const Value.absent()
+              : Value(serverUploadId),
+          serverPhotoId: serverPhotoId == null
+              ? const Value.absent()
+              : Value(serverPhotoId),
+          bytes: bytes == null ? const Value.absent() : Value(bytes),
           updatedAt: Value(_utcAt(now)),
         ),
       );

apps/mobile/lib/features/uploads/upload_engine.dart

@@ -0,0 +1,602 @@
+/// 上传传输引擎 + 源文件清理执行器（P3 计划 Task 6）。
+///
+/// 架构（对齐技术实施方案 §5.5）：
+/// - **驱动**：周期 timer + 工作可用通知（[notifyWorkAvailable]，准备器
+///   入队时挂接）双触发；每次唤醒执行一轮调度（[drain]）——补足并行
+///   配额（[kMaxParallelUploads]，dart-define MAX_PARALLEL_UPLOADS 覆盖）；
+/// - **出队**：账号隔离（TaskStore 读路径硬守护）内取最旧就绪任务：
+///   queued 恒就绪；waitingNetwork/retryableFailed 需 nextRetryAt 到期
+///   （retryableFailed 的 nextRetryAt=null 表示自动重试上限已到/需人工
+///   介入，永不自动出队）；网络恢复（[onNetworkRestored]）对
+///   waitingNetwork 立即绕过退避；
+/// - **传输三步**：initUpload（幂等，clientTaskId + sourceDeviceId）→
+///   presign PUT（dio 流式 + 进度回调 + 取消令牌；**裸 Dio**——不挂
+///   ApiClient 拦截器，避免 Bearer 头污染 MinIO 预签名签名）→
+///   completeUpload（幂等核验）→ completed；
+/// - **失败分类**（[UploadEngineSourceMissing] 之外见 [_classifyFailure]）：
+///   | 错误 | 目标状态 | 自动重试 |
+///   |---|---|---|
+///   | 网络不可达/超时/429 限速 | waitingNetwork | 退避 5s·2^n 封顶 10min，无次数上限 |
+///   | 401 会话过期 | waitingNetwork | 30s 短退避；全局驱逐由 ApiClient 处理，不烧次数 |
+///   | 400 UPLOAD_VERIFY_FAILED / 5xx / 未知 | retryableFailed | 退避重试，3 次后停止自动重试 |
+///   | 409 ALBUM_UPLOAD_DISABLED / UPLOAD_SESSION_CONFLICT / 源文件缺失 | retryableFailed | 不自动重试（人工修正后 retryNow） |
+///   | presign 403 凭证过期 | （renew 后原状重传） | renewUpload 至多 2 次 |
+/// - **取消**（[cancel]）：未开始 → cancelled 后删行删文件（确定语义）；
+///   进行中 → cancelToken 尽力取消 + 置 cancelled（文件保留，由清理
+///   执行器按保留期回收）；
+/// - **崩溃恢复**：进程被杀时卡在 uploading/uploadedConfirming 且不在
+///   in-flight 表内的任务，每轮 drain 前归位 waitingNetwork 立即重试
+///   （initUpload/completeUpload 幂等，重传安全；A05 模拟器级在 Task 11
+///   集成验证）；
+/// - **A15 连拍归属不变**：引擎只读任务行的 albumId/categoryId（无任何
+///   写通道），入队后相册/分类切换不影响既有任务；
+/// - **清理执行器**（[cleanup]，随每轮 drain 与 App 启动执行）：仅终态
+///   （completed/cancelled，TaskStore 硬守护——未完成源文件永不清理）且
+///   超过 KEEP_COMPLETED_DAYS 保留期的源文件删除 + cleanedAt 标记
+///   （行保留作记录）。
+library;
+
+import 'dart:async';
+import 'dart:io';
+
+import 'package:dio/dio.dart';
+
+import '../../core/api_client.dart';
+import '../../core/app_exception.dart';
+import '../../core/device_id.dart';
+import '../../core/error_codes.dart';
+import 'models.dart';
+import 'task_state.dart';
+import 'task_store.dart';
+
+/// 并行传输上限（计划：批量并发为可配置小值；dart-define
+/// MAX_PARALLEL_UPLOADS 覆盖）。
+const int kMaxParallelUploads =
+    int.fromEnvironment('MAX_PARALLEL_UPLOADS', defaultValue: 2);
+
+/// completed/cancelled 任务源文件保留天数（dart-define
+/// KEEP_COMPLETED_DAYS 覆盖）。
+const int kKeepCompletedDays =
+    int.fromEnvironment('KEEP_COMPLETED_DAYS', defaultValue: 14);
+
+/// 网络类失败的基础退避（5s·2^(attempt-1)，封顶 10min）。
+const Duration kBackoffBase = Duration(seconds: 5);
+
+/// 退避封顶。
+const Duration kBackoffCap = Duration(minutes: 10);
+
+/// retryableFailed 自动重试次数上限（UPLOAD_VERIFY_FAILED 等；到达后
+/// 停止自动重试，等人工 [UploadEngine.retryNow]）。
+const int kMaxAutoRetryAttempts = 3;
+
+/// presign 凭证过期（403）的最多元数（renewUpload 次数）。
+const int kMaxCredentialRenewals = 2;
+
+/// 传输引擎。
+///
+/// 事件驱动 + 周期兜底；不 own isolate（传输多为 IO 等待，UI isolate
+/// 足够；Android 后台延续由 Task 7 原生桥接承担）。
+class UploadEngine {
+  UploadEngine(
+    this._store,
+    this._apiClient,
+    this._deviceId, {
+    Dio? objectDio,
+    DateTime Function()? now,
+    this.maxParallel = kMaxParallelUploads,
+    this.keepCompletedDays = kKeepCompletedDays,
+    this.onProgress,
+    this.onError,
+  })  : _objectDio = objectDio ?? Dio(),
+        _now = now ?? DateTime.now;
+
+  final TaskStore _store;
+  final ApiClient _apiClient;
+  final DeviceId _deviceId;
+
+  /// presign PUT 专用裸 Dio（无鉴权拦截器，见库级注释）。
+  final Dio _objectDio;
+  final DateTime Function() _now;
+
+  /// 并行传输上限（运行时 <1 钳位为 1，见 [_effectiveParallel]）。
+  final int maxParallel;
+
+  int get _effectiveParallel => maxParallel < 1 ? 1 : maxParallel;
+
+  /// 源文件保留天数（清理执行器）。
+  final int keepCompletedDays;
+
+  /// 传输进度回调（taskId, 已发字节, 总字节；队列 UI 进度条挂接点）。
+  final void Function(String taskId, int sent, int total)? onProgress;
+
+  /// 引擎级异常观察（未预期中断性错误；分类内失败不经此处）。
+  final void Function(String taskId, Object error)? onError;
+
+  /// 周期唤醒 timer（[start] 创建，[stop] 取消；测试不 start，直接
+  /// 驱动 [drain]/[settle]）。
+  Timer? _timer;
+
+  /// 当前引擎服务的账号（出队范围；账号切换=重新 [start]）。
+  String? _accountId;
+
+  /// in-flight 任务表（崩溃恢复的「不在表内」判据 + 防同任务双发）。
+  final Map<String, _TransferHandle> _inFlight = <String, _TransferHandle>{};
+
+  /// drain 进行中Future（并发调用折叠为同轮——所有 await 方都等到完整
+  /// 调度结束；结束后清空）。
+  Future<void>? _drainPending;
+
+  /// 网络恢复信号：下一轮 drain 对 waitingNetwork 绕过退避（一次性）。
+  bool _bypassBackoffOnce = false;
+
+  /// 源文件缺失伪码（客户端本地；受管副本被系统清理等，不可恢复）。
+  static const String sourceMissingCode = 'UPLOAD_SOURCE_MISSING';
+
+  /// 启动引擎（账号切换时重复调用即换 scope；在途传输不被中断，自然
+  /// 收尾后不再出队旧账号任务）。
+  void start(String accountId,
+      {Duration period = const Duration(seconds: 15)}) {
+    _accountId = accountId;
+    _timer?.cancel();
+    _timer = Timer.periodic(period, (_) => drain());
+    unawaited(drain());
+  }
+
+  /// 停止引擎（登出/退后台策略；不中断在途传输，仅停止再出队）。
+  void stop() {
+    _timer?.cancel();
+    _timer = null;
+    _accountId = null;
+  }
+
+  /// 工作可用通知（准备器入队回调挂接）：尽快调度一轮。
+  void notifyWorkAvailable() => unawaited(drain());
+
+  /// 网络恢复通知（连接监听挂接）：waitingNetwork 立即绕过退避重试。
+  void onNetworkRestored() {
+    _bypassBackoffOnce = true;
+    unawaited(drain());
+  }
+
+  /// 是否仍在运行（周期 timer 挂着）。
+  bool get isRunning => _timer != null;
+
+  /// 等待全部在途传输结束（测试与优雅停机用）。
+  Future<void> settle() => Future.wait(
+        [for (final handle in _inFlight.values) handle.done],
+      ).then((_) {});
+
+  /// 一轮调度：崩溃恢复 → 补足并行配额。并发调用折叠为同轮（见
+  /// [_drainPending]），所有 await 方等到完整调度结束。
+  Future<void> drain() {
+    final pending = _drainPending;
+    if (pending != null) return pending;
+    final future = _doDrain().whenComplete(() => _drainPending = null);
+    _drainPending = future;
+    return future;
+  }
+
+  Future<void> _doDrain() async {
+    final accountId = _accountId;
+    if (accountId == null) return;
+    final bypass = _bypassBackoffOnce;
+    _bypassBackoffOnce = false;
+    await _recoverInterrupted(accountId);
+    while (_inFlight.length < _effectiveParallel) {
+      final task = await _pickReady(accountId, bypassBackoff: bypass);
+      if (task == null) break;
+      _launch(task);
+    }
+  }
+
+  /// 就绪任务挑选（账号内 createdAt 最旧优先——store 已排序）。
+  Future<UploadTask?> _pickReady(
+    String accountId, {
+    bool bypassBackoff = false,
+  }) async {
+    final candidates = await _store.listByAccount(
+      accountId,
+      states: const {
+        UploadTaskState.queued,
+        UploadTaskState.waitingNetwork,
+        UploadTaskState.retryableFailed,
+      },
+    );
+    final now = _now();
+    for (final task in candidates) {
+      if (_inFlight.containsKey(task.id)) continue;
+      switch (task.state) {
+        case UploadTaskState.queued:
+          return task;
+        case UploadTaskState.waitingNetwork:
+          // 网络恢复绕过退避（一次性，仅此状态）。
+          if (bypassBackoff || _retryDue(task.nextRetryAt, now)) {
+            return task;
+          }
+        case UploadTaskState.retryableFailed:
+          // nextRetryAt=null：自动重试上限已到，等人工 retryNow。
+          if (_retryDue(task.nextRetryAt, now)) {
+            return task;
+          }
+        case UploadTaskState.preparing:
+        case UploadTaskState.preparationFailed:
+        case UploadTaskState.uploading:
+        case UploadTaskState.uploadedConfirming:
+        case UploadTaskState.completed:
+        case UploadTaskState.cancelled:
+          break; // 不在候选集（防御性穷举）。
+      }
+    }
+    return null;
+  }
+
+  static bool _retryDue(DateTime? nextRetryAt, DateTime now) =>
+      nextRetryAt != null && !nextRetryAt.isAfter(now);
+
+  /// 崩溃恢复：uploading/uploadedConfirming 但不在 in-flight 表 →
+  /// waitingNetwork（nextRetryAt=now 立即重试；幂等保障安全）。
+  Future<void> _recoverInterrupted(String accountId) async {
+    final stuck = await _store.listByAccount(
+      accountId,
+      states: const {
+        UploadTaskState.uploading,
+        UploadTaskState.uploadedConfirming,
+      },
+    );
+    for (final task in stuck) {
+      if (_inFlight.containsKey(task.id)) continue;
+      try {
+        await _store.updateState(
+          task.id,
+          UploadTaskState.waitingNetwork,
+          patch: UploadTaskPatch(nextRetryAt: _now()),
+          now: _now(),
+        );
+      } on StateError {
+        // 并发迁移（取消等）——恢复放弃该任务本轮处理。
+      }
+    }
+  }
+
+  /// 启动一次传输：登记 in-flight（同步，防双发）后转入后台协程。
+  void _launch(UploadTask task) {
+    final handle = _TransferHandle();
+    _inFlight[task.id] = handle;
+    final future = _runTransfer(task, handle);
+    handle.done = future;
+    unawaited(
+      future.whenComplete(() {
+        _inFlight.remove(task.id);
+        // 完成即补位（可能仍有就绪工作）。
+        unawaited(drain());
+      }),
+    );
+  }
+
+  /// 单任务传输协程：状态入 uploading → init → PUT（含 renew）→
+  /// complete，失败全程分类落库。
+  Future<void> _runTransfer(UploadTask task, _TransferHandle handle) async {
+    // retryableFailed → queued → uploading（状态机需经 queued 中转）；
+    // waitingNetwork/queued → uploading 直达。
+    try {
+      var current = task;
+      if (current.state == UploadTaskState.retryableFailed) {
+        current = await _store.updateState(
+          current.id,
+          UploadTaskState.queued,
+          now: _now(),
+        );
+      }
+      await _store.updateState(
+        current.id,
+        UploadTaskState.uploading,
+        patch: const UploadTaskPatch(clearNextRetryAt: true),
+        now: _now(),
+      );
+      await _transferSteps(current, handle);
+    } on DioException catch (e) {
+      // 仅来自 PUT（api 调用已归一 AppException）；取消已在
+      // _transferSteps 内短路，到达这里的均为待分类失败。
+      await _classifyFailure(task, e);
+    } on AppException catch (e) {
+      await _classifyFailure(task, e);
+    } on StateError {
+      // 并发取消抢走了状态迁移（任务已 cancelled）：尽力取消语义达成。
+    } on Object catch (e) {
+      // 未预期错误（文件系统等）：按可重试失败兜底 + 观察。
+      onError?.call(task.id, e);
+      await _classifyFailure(
+        task,
+        AppException(code: ErrorCodes.internalError, message: 'engine: $e'),
+      );
+    }
+  }
+
+  Future<void> _transferSteps(UploadTask task, _TransferHandle handle) async {
+    // a. initUpload（幂等：重试复用 clientTaskId，服务端返回原会话）。
+    final init = await _apiClient.initUpload(
+      UploadInitRequest(
+        albumId: task.albumId,
+        categoryId: task.categoryId,
+        clientTaskId: task.clientTaskId,
+        filename: task.filename,
+        bytes: task.bytes,
+        contentType: task.contentType,
+        sourceDeviceId: await _deviceId.value(),
+      ),
+    );
+    // 回填会话/照片 ID（状态仍 uploading，旁路写入，见 TaskStore.updateMeta）。
+    await _store.updateMeta(
+      task.id,
+      serverUploadId: init.uploadId,
+      serverPhotoId: init.photoId,
+      now: _now(),
+    );
+
+    // b. presign PUT（流式 + 进度 + 取消令牌；403 → renew 换凭证重传）。
+    var credential = init.credential;
+    var renewals = 0;
+    while (true) {
+      try {
+        await _putObject(task, credential, handle.cancelToken);
+        break;
+      } on DioException catch (e) {
+        if (handle.cancelToken.isCancelled) {
+          // 用户取消：状态已置 cancelled（cancel() 内），终止协程。
+          return;
+        }
+        if (e.response?.statusCode == 403 &&
+            renewals < kMaxCredentialRenewals) {
+          renewals++;
+          final renewed = await _apiClient.renewUpload(init.uploadId);
+          credential = renewed.credential;
+          continue;
+        }
+        rethrow;
+      }
+    }
+
+    // c. completeUpload（幂等核验）→ completed。
+    await _store.updateState(
+      task.id,
+      UploadTaskState.uploadedConfirming,
+      now: _now(),
+    );
+    final complete = await _apiClient.completeUpload(init.uploadId);
+    await _store.updateState(
+      task.id,
+      UploadTaskState.completed,
+      patch: UploadTaskPatch(
+        serverPhotoId: complete.photoId,
+        clearLastError: true,
+        clearNextRetryAt: true,
+      ),
+      now: _now(),
+    );
+  }
+
+  /// presign PUT：裸 Dio 直传 credential.url，严格携带 credential.headers
+  /// （+ Content-Length——流式发送 dio 无法自动推断，显式声明）。进度经
+  /// 计数流上报（引擎自持，不依赖适配器对 onSendProgress 的实现）。
+  Future<void> _putObject(
+    UploadTask task,
+    UploadCredential credential,
+    CancelToken cancelToken,
+  ) async {
+    final file = File(task.sourceFilePath);
+    if (!await file.exists()) {
+      // 源文件缺失：不可恢复，分类按「不自动重试」处理。
+      throw const AppException(
+        code: UploadEngine.sourceMissingCode,
+        message: 'source file missing',
+      );
+    }
+    final length = await file.length();
+    var sent = 0;
+    final progressStream =
+        file.openRead().map((List<int> chunk) {
+      sent += chunk.length;
+      onProgress?.call(task.id, sent, length);
+      return chunk;
+    });
+    await _objectDio.put<void>(
+      credential.url,
+      data: progressStream,
+      cancelToken: cancelToken,
+      options: Options(
+        method: credential.method,
+        headers: {
+          ...credential.headers,
+          Headers.contentLengthHeader: length,
+        },
+        // 响应体非 JSON（MinIO XML/空），防 transformer 解析报错。
+        responseType: ResponseType.plain,
+        receiveDataWhenStatusError: true,
+      ),
+    );
+  }
+
+  /// 失败分类与状态落库（分类表见库级注释；attempt 每次 +1，目标状态
+  /// 自 uploading/uploadedConfirming 均合法迁移）。
+  Future<void> _classifyFailure(UploadTask task, Object error) async {
+    final AppException app;
+    if (error is AppException) {
+      app = error;
+    } else if (error is DioException) {
+      app = ApiClient.mapDioException(error);
+    } else {
+      app = AppException(code: ErrorCodes.internalError, message: '$error');
+    }
+    final code = app.code;
+    final status = app.statusCode;
+    final attempt = task.attempt + 1;
+
+    // 1) 网络类（含超时）与限速：waitingNetwork 退避（无次数上限）。
+    if (code == ErrorCodes.networkError ||
+        code == ErrorCodes.networkTimeout ||
+        code == ErrorCodes.rateLimited) {
+      await _fail(
+        task,
+        UploadTaskState.waitingNetwork,
+        UploadTaskPatch(
+          attempt: attempt,
+          nextRetryAt: _now().add(backoffFor(attempt)),
+          lastError: code,
+        ),
+      );
+      return;
+    }
+
+    // 2) 会话过期：waitingNetwork 短退避（重登录后续传；全局驱逐由
+    //    ApiClient.onUnauthorized 处理，不烧自动重试次数、attempt 不变）。
+    if (status == 401 || code == ErrorCodes.authTokenExpired) {
+      await _fail(
+        task,
+        UploadTaskState.waitingNetwork,
+        UploadTaskPatch(
+          nextRetryAt: _now().add(const Duration(seconds: 30)),
+          lastError: code,
+        ),
+      );
+      return;
+    }
+
+    // 3) 客户端重试无意义（服务端开关/幂等冲突/源文件缺失）：不自动
+    //    重试，用户可见 lastError，修正后 retryNow 手动重试。
+    if (code == ErrorCodes.albumUploadDisabled ||
+        code == ErrorCodes.uploadSessionConflict ||
+        code == UploadEngine.sourceMissingCode) {
+      await _fail(
+        task,
+        UploadTaskState.retryableFailed,
+        UploadTaskPatch(lastError: code),
+      );
+      return;
+    }
+
+    // 4) 其余（UPLOAD_VERIFY_FAILED / 5xx / 未知）：retryableFailed 退避
+    //    自动重试，达上限后停止（nextRetryAt 缺省=null）。
+    final autoRetry = attempt < kMaxAutoRetryAttempts;
+    await _fail(
+      task,
+      UploadTaskState.retryableFailed,
+      UploadTaskPatch(
+        attempt: attempt,
+        nextRetryAt: autoRetry ? _now().add(backoffFor(attempt)) : null,
+        lastError: code,
+      ),
+    );
+  }
+
+  /// 分类落地（并发取消致 StateError 时静默——cancelled 已是终态）。
+  Future<void> _fail(
+    UploadTask task,
+    UploadTaskState to,
+    UploadTaskPatch patch,
+  ) async {
+    try {
+      await _store.updateState(task.id, to, patch: patch, now: _now());
+    } on StateError {
+      // 已被并发取消（cancelled 终态）：保留取消语义。
+    }
+  }
+
+  /// 退避时长：5s·2^(attempt-1)，封顶 10min（attempt ≥ 1）。
+  ///
+  /// 指数先钳位：Dart VM 下 `1 << n`（n ≥ 64）归零，超大 attempt 会使
+  /// 乘数溢出为 0；且 2^7 档（640s）已越过封顶，更大指数无意义。
+  static Duration backoffFor(int attempt) {
+    if (attempt < 1) return kBackoffBase;
+    const maxShift = 7;
+    final shift = attempt - 1 < maxShift ? attempt - 1 : maxShift;
+    final shifted = kBackoffBase * (1 << shift);
+    return shifted > kBackoffCap ? kBackoffCap : shifted;
+  }
+
+  /// 取消任务（语义见库级注释）。返回是否生效（终态/不存在=false）。
+  Future<bool> cancel(String accountId, String taskId) async {
+    final task = await _store.get(accountId, taskId);
+    if (task == null || task.state.isTerminal) return false;
+    await _store.updateState(
+      taskId,
+      UploadTaskState.cancelled,
+      patch: const UploadTaskPatch(clearNextRetryAt: true),
+      now: _now(),
+    );
+    final handle = _inFlight[taskId];
+    if (handle != null) {
+      // 进行中：尽力取消（协程在取消异常/下一检查点收敛）。
+      handle.cancelToken.cancel();
+      return true;
+    }
+    // 未开始：删行 + 删受管文件（best-effort；失败不回滚取消）。
+    await _store.delete(taskId);
+    try {
+      final file = File(task.sourceFilePath);
+      if (await file.exists()) {
+        await file.delete();
+      }
+    } on Object {
+      // 残留文件由清理执行器路径之外的孤儿——接受（行已删，无清理候选）。
+    }
+    return true;
+  }
+
+  /// 手动重试（队列 UI「重试」按钮）：retryableFailed（含达上限）/
+  /// waitingNetwork 任务立即重新入队出队。
+  Future<void> retryNow(String accountId, String taskId) async {
+    final task = await _store.get(accountId, taskId);
+    if (task == null) return;
+    if (task.state != UploadTaskState.retryableFailed &&
+        task.state != UploadTaskState.waitingNetwork) {
+      return;
+    }
+    await _store.updateState(
+      taskId,
+      UploadTaskState.queued,
+      patch: const UploadTaskPatch(clearNextRetryAt: true),
+      now: _now(),
+    );
+    unawaited(drain());
+  }
+
+  /// 清理执行器：删除超过保留期的终态任务源文件并标记 cleanedAt（行
+  /// 保留作历史记录）。先删文件后标记——中断重入安全（文件已删为
+  /// no-op）。返回清理条数。候选集硬守护在 TaskStore.cleanupCandidates
+  /// （未完成任务永不出现）。
+  Future<int> cleanup({DateTime? now}) async {
+    final at = now ?? _now();
+    final candidates = await _store.cleanupCandidates(keepCompletedDays, now: at);
+    var cleaned = 0;
+    for (final task in candidates) {
+      try {
+        final file = File(task.sourceFilePath);
+        if (await file.exists()) {
+          await file.delete();
+        }
+      } on Object {
+        continue; // 单文件失败不阻断其余（下次唤醒重试）。
+      }
+      await _store.markCleanable(task.id, now: at);
+      cleaned++;
+    }
+    return cleaned;
+  }
+
+  /// 关闭（App 退出）：停 timer、尽力断在途、释放 object Dio。
+  void dispose() {
+    stop();
+    for (final handle in _inFlight.values) {
+      handle.cancelToken.cancel();
+    }
+    _objectDio.close();
+  }
+}
+
+/// 在途传输控制句柄。
+class _TransferHandle {
+  final CancelToken cancelToken = CancelToken();
+
+  /// 传输协程完成信号（settle 用；_launch 内同步赋值）。
+  Future<void> done = Future<void>.value();
+}