fix: 修复 flake8 F541 f-string 无占位符

移除 generation.py:452 处多余的 f 前缀，消除 CI lint 报错。

alembic/versions/037_add_logs_to_generation_tasks.py

@@ -0,0 +1,26 @@
+"""Add logs field to generation_tasks
+
+Revision ID: 037_generation_logs
+Revises: 036_expand_uuid_36
+Create Date: 2026-07-10
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "037_generation_logs"
+down_revision = "036_expand_uuid_36"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "generation_tasks",
+        sa.Column("logs", sa.Text(), nullable=False, server_default="[]"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("generation_tasks", "logs")

apps/api/app/api/routes/generation_tasks.py

@@ -1,3 +1,4 @@
+import logging
 import random
 import uuid
 from typing import Any
@@ -31,6 +32,8 @@ from packages.application import (
     ListGeneratedVideosByTaskUseCase,
 )
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter()
 
 
@@ -57,6 +60,7 @@ def _to_generation_task_response(task) -> GenerationTaskResponse:
         source_edit_plan_id=task.source_edit_plan_id or "",
         asset_select_mode=getattr(task, "asset_select_mode", ""),
         batch_id=getattr(task, "batch_id", ""),
+        logs=getattr(task, "logs", "[]"),
         status=task.status,
         progress=task.progress,
         result_count=task.result_count,
@@ -181,19 +185,37 @@ def create_generation_task(
     asset_library_repository: Any = Depends(get_asset_library_repository),
     asset_repository: Any = Depends(get_asset_repository),
 ) -> BatchGenerationTaskResponse:
-    project_id, asset_library_id = _resolve_project_and_library(
-        request, project_repository, asset_library_repository, asset_repository, authenticated_user
+    logger.info(
+        "[生成任务] 接收请求: user_id=%s, template_id=%s, asset_count=%d, mode=%s, count=%d",
+        authenticated_user.user.id,
+        request.template_id,
+        len(request.asset_ids),
+        request.asset_select_mode,
+        request.count,
     )
 
+    try:
+        project_id, asset_library_id = _resolve_project_and_library(
+            request, project_repository, asset_library_repository, asset_repository, authenticated_user
+        )
+    except HTTPException as e:
+        logger.warning("[生成任务] 校验失败: %s", e.detail)
+        raise
+
     # asset_library 存在性校验（仅在提供了 asset_library_id 时）
     resolved_asset_ids: list[str] = list(request.asset_ids)
     if asset_library_id:
         library = asset_library_repository.get(asset_library_id)
         if library is None or (project_id and library.project_id != project_id):
+            logger.warning("[生成任务] 素材库不存在: library_id=%s", asset_library_id)
             raise HTTPException(status_code=404, detail=f"AssetLibrary {asset_library_id} not found")
 
         assets = asset_repository.find_by_library(asset_library_id)
-        _ensure_library_has_ready_video_assets(assets)
+        try:
+            _ensure_library_has_ready_video_assets(assets)
+        except HTTPException as e:
+            logger.warning("[生成任务] 素材校验失败: %s", e.detail)
+            raise
 
         # 素材库自动匹配：当未显式指定 asset_ids 时，按模式自动选取
         if not resolved_asset_ids:
@@ -209,25 +231,35 @@ def create_generation_task(
     # 同批次任务共享 batch_id，用于视频查重时批次内比对
     batch_id = uuid.uuid4().hex if count > 1 else ""
 
-    for _ in range(count):
-        task = use_case.execute(
-            CreateGenerationTaskCommand(
-                project_id=project_id,
-                asset_library_id=asset_library_id,
-                strategy_id=request.strategy_id,
-                voice_library_id=request.voice_library_id,
-                template_id=request.template_id,
-                asset_ids=resolved_asset_ids,
-                title_ids=request.title_ids,
-                voice_ids=request.voice_ids,
-                created_by_user_id=authenticated_user.user.id,
-                source_edit_plan_id=request.source_edit_plan_id,
-                asset_select_mode=request.asset_select_mode,
-                batch_id=batch_id,
+    try:
+        for _ in range(count):
+            task = use_case.execute(
+                CreateGenerationTaskCommand(
+                    project_id=project_id,
+                    asset_library_id=asset_library_id,
+                    strategy_id=request.strategy_id,
+                    voice_library_id=request.voice_library_id,
+                    template_id=request.template_id,
+                    asset_ids=resolved_asset_ids,
+                    title_ids=request.title_ids,
+                    voice_ids=request.voice_ids,
+                    created_by_user_id=authenticated_user.user.id,
+                    source_edit_plan_id=request.source_edit_plan_id,
+                    asset_select_mode=request.asset_select_mode,
+                    batch_id=batch_id,
+                )
             )
-        )
-        celery_app.send_task("worker.generate_video", args=[task.id])
-        created_tasks.append(task)
+            celery_app.send_task("worker.generate_video", args=[task.id])
+            created_tasks.append(task)
+            logger.info(
+                "[生成任务] 入队成功: task_id=%s, status=%s, batch_id=%s",
+                task.id,
+                task.status,
+                batch_id,
+            )
+    except Exception as e:
+        logger.error("[生成任务] 创建失败: %s", e, exc_info=True)
+        raise HTTPException(status_code=500, detail="创建生成任务失败，请稍后重试或查看任务日志")
 
     items = [_to_generation_task_response(t) for t in created_tasks]
     return BatchGenerationTaskResponse(items=items, total=len(items))

apps/api/app/schemas/generation_task.py

@@ -1,4 +1,6 @@
-from pydantic import BaseModel, Field, model_validator
+import json
+
+from pydantic import BaseModel, Field, field_validator, model_validator
 
 
 class CreateGenerationTaskRequest(BaseModel):
@@ -62,6 +64,21 @@ class GenerationTaskResponse(BaseModel):
     progress: float
     result_count: int
     error_message: str
+    logs: list[dict] = Field(default_factory=list)
+
+    @field_validator("logs", mode="before")
+    @classmethod
+    def _parse_logs(cls, v: object) -> list[dict]:
+        """将 JSON 字符串解析为 list[dict]。"""
+        if isinstance(v, str):
+            try:
+                parsed = json.loads(v)
+                return parsed if isinstance(parsed, list) else []
+            except (json.JSONDecodeError, TypeError):
+                return []
+        if isinstance(v, list):
+            return v
+        return []
 
 
 class BatchGenerationTaskResponse(BaseModel):

apps/worker/worker_app/tasks/generation.py

@@ -13,10 +13,13 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import os
 import tempfile
+import time
 from dataclasses import dataclass, field
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Optional
 
@@ -81,6 +84,26 @@ def _update_task_status(task_id: str, status_action: str, **kwargs) -> bool:
         return False
 
 
+# ── 日志持久化辅助 ────────────────────────────────────────────────────────────
+
+
+def _flush_logs(task_id: str, gen_task) -> None:
+    """将 gen_task.logs 持久化到 DB（独立 session，失败不抛异常）。"""
+    try:
+        session = SessionLocal()
+        try:
+            from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
+
+            model = session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first()
+            if model:
+                model.logs = gen_task.logs
+                session.commit()
+        finally:
+            session.close()
+    except Exception:
+        logger.warning("[task_id=%s] 日志持久化失败", task_id, exc_info=True)
+
+
 # ── 共享工具模块导入 ──────────────────────────────────────────────────────────
 
 from video_processing.dedup_helpers import create_video_record_and_dedup
@@ -320,6 +343,8 @@ def _download_library_assets(
     asset_ids: list[str] | None = None,
     video_extensions: tuple = (".mp4", ".mov", ".avi", ".mkv", ".webm"),
     strict: bool = True,
+    task_id: str = "",
+    gen_task=None,
 ) -> list[Path]:
     """下载视频素材 — 同时支持素材库模式和项目级模式。
 
@@ -419,19 +444,65 @@ def _download_library_assets(
                 storage_key = asset.file_url if asset.file_url else None
                 if not storage_key:
                     failed_assets.append(f"{asset.name}({asset.id})")
-                    logger.warning("素材缺少 file_url, 跳过: asset_id=%s name=%s", asset.id, asset.name)
+                    logger.warning(
+                        "[task_id=%s] 素材缺少 file_url, 跳过: asset_id=%s name=%s", task_id, asset.id, asset.name
+                    )
+                    if gen_task:
+                        gen_task.append_log(
+                            "下载素材",
+                            "素材缺少file_url, 跳过",
+                            level="WARN",
+                            asset_id=asset.id,
+                            asset_name=asset.name,
+                            success=False,
+                            file_size=0,
+                            duration=0.0,
+                        )
                     if strict:
                         raise RuntimeError(f"素材缺少 file_url: asset_id={asset.id}, name={asset.name}")
                     continue
 
                 ext = Path(storage_key).suffix or ".mp4"
                 local_file = temp_path / f"asset_{i:03d}_{asset.id}{ext}"
-                if download_asset(storage_key, local_file):
+                asset_start = time.monotonic()
+                download_ok = download_asset(storage_key, local_file)
+                asset_elapsed = time.monotonic() - asset_start
+
+                if download_ok:
+                    file_size = local_file.stat().st_size if local_file.exists() else 0
                     downloaded.append(local_file)
-                    logger.info("Downloaded asset: %s -> %s", asset.name, local_file)
+                    logger.info(
+                        "[task_id=%s] Downloaded asset: %s -> %s (size=%d, time=%.1fs)",
+                        task_id,
+                        asset.name,
+                        local_file,
+                        file_size,
+                        asset_elapsed,
+                    )
+                    if gen_task:
+                        gen_task.append_log(
+                            "下载素材",
+                            f"下载成功: {asset.name}",
+                            asset_id=asset.id,
+                            asset_name=asset.name,
+                            success=True,
+                            file_size=file_size,
+                            duration=round(asset_elapsed, 2),
+                        )
                 else:
                     failed_assets.append(f"{asset.name}({asset.id})")
-                    logger.warning("Failed to download asset: %s (id=%s)", asset.name, asset.id)
+                    logger.warning("[task_id=%s] Failed to download asset: %s (id=%s)", task_id, asset.name, asset.id)
+                    if gen_task:
+                        gen_task.append_log(
+                            "下载素材",
+                            f"下载失败: {asset.name}",
+                            level="WARN",
+                            asset_id=asset.id,
+                            asset_name=asset.name,
+                            success=False,
+                            file_size=0,
+                            duration=round(asset_elapsed, 2),
+                        )
                     if strict:
                         raise RuntimeError(f"素材下载失败: asset_id={asset.id}, name={asset.name}")
 
@@ -511,7 +582,7 @@ def generate_video(self, task_id: str) -> dict:
     """
     from packages.domain import EditingMode
 
-    logger.info("开始生成视频任务: task_id=%s", task_id)
+    logger.info("[task_id=%s] [接收任务] 开始生成视频任务", task_id)
 
     # 从数据库加载任务信息
     session = SessionLocal()
@@ -523,7 +594,7 @@ def generate_video(self, task_id: str) -> dict:
         task_repo = SQLAlchemyGenerationTaskRepository(session)
         gen_task = task_repo.get(task_id)
         if gen_task is None:
-            logger.error("生成任务不存在: task_id=%s", task_id)
+            logger.error("[task_id=%s] [接收任务] 任务不存在", task_id)
             return {"status": "failed", "error": f"generation task {task_id} not found"}
         project_id = gen_task.project_id
         asset_library_id = gen_task.asset_library_id
@@ -532,6 +603,16 @@ def generate_video(self, task_id: str) -> dict:
         mode = gen_task.strategy_id or "one_take"
         task_asset_ids = list(gen_task.asset_ids or [])
         batch_id = getattr(gen_task, "batch_id", "") or ""
+
+        # 记录接收任务日志
+        gen_task.append_log(
+            "接收任务",
+            f"模式={mode}, 模板={template_id}, 素材数={len(task_asset_ids)}",
+            mode=mode,
+            template_id=template_id,
+            asset_count=len(task_asset_ids),
+        )
+        _flush_logs(task_id, gen_task)
     finally:
         session.close()
 
@@ -558,12 +639,40 @@ def generate_video(self, task_id: str) -> dict:
             output_path = temp_path / output_name
 
             # 1. 从素材库/项目下载视频素材
+            logger.info("[task_id=%s] [下载素材] 开始下载视频素材", task_id)
+            download_start = time.monotonic()
             downloaded_videos = _download_library_assets(
                 temp_path,
                 asset_library_id=asset_library_id,
                 project_id=project_id,
                 asset_ids=task_asset_ids or None,
+                task_id=task_id,
+                gen_task=gen_task,
             )
+            download_elapsed = time.monotonic() - download_start
+            logger.info(
+                "[task_id=%s] [下载素材] 完成: 成功=%d个, 耗时=%.1fs",
+                task_id,
+                len(downloaded_videos),
+                download_elapsed,
+            )
+
+            # 重新加载 gen_task 以追加日志（session 已关闭）
+            _session = SessionLocal()
+            try:
+                _repo = SQLAlchemyGenerationTaskRepository(_session)
+                gen_task = _repo.get(task_id)
+            finally:
+                _session.close()
+
+            if gen_task:
+                gen_task.append_log(
+                    "下载素材",
+                    f"成功下载 {len(downloaded_videos)} 个视频素材",
+                    count=len(downloaded_videos),
+                    duration=round(download_elapsed, 2),
+                )
+                _flush_logs(task_id, gen_task)
 
             # 2. 下载配音（如有）
             audio_path: str | None = None
@@ -571,6 +680,7 @@ def generate_video(self, task_id: str) -> dict:
                 local_audio = temp_path / "voice.mp3"
                 if _download_voice_asset(voice_library_id, local_audio):
                     audio_path = str(local_audio)
+                    logger.info("[task_id=%s] [下载配音] 配音下载成功", task_id)
 
             # 3. 渲染
             if not downloaded_videos:
@@ -588,7 +698,26 @@ def generate_video(self, task_id: str) -> dict:
                 mode=editing_mode.value,
             )
 
+            total_duration = sum(c.duration for c in virtual_clips)
+            logger.info(
+                "[task_id=%s] [剪辑计划] 片段数=%d, 总时长=%.1fs",
+                task_id,
+                len(virtual_clips),
+                total_duration,
+            )
+
+            if gen_task:
+                gen_task.append_log(
+                    "剪辑计划",
+                    f"片段数={len(virtual_clips)}, 总时长={total_duration:.1f}s",
+                    segment_count=len(virtual_clips),
+                    total_duration=round(total_duration, 2),
+                )
+                _flush_logs(task_id, gen_task)
+
             # 使用 UnifiedRenderService 渲染
+            logger.info("[task_id=%s] [渲染] FFmpeg 渲染开始", task_id)
+            render_start = time.monotonic()
             render_service = UnifiedRenderService(
                 plan=virtual_plan,
                 clips=virtual_clips,
@@ -599,6 +728,20 @@ def generate_video(self, task_id: str) -> dict:
                 output_fps=int(OUTPUT_FPS),
             )
             render_result = render_service.render()
+            render_elapsed = time.monotonic() - render_start
+            logger.info(
+                "[task_id=%s] [渲染] FFmpeg 渲染完成: 耗时=%.1fs",
+                task_id,
+                render_elapsed,
+            )
+
+            if gen_task:
+                gen_task.append_log(
+                    "渲染",
+                    f"FFmpeg 渲染完成, 耗时={render_elapsed:.1f}s",
+                    duration=round(render_elapsed, 2),
+                )
+                _flush_logs(task_id, gen_task)
 
             # 4. 如有配音，后处理混音
             if audio_path:
@@ -608,7 +751,7 @@ def generate_video(self, task_id: str) -> dict:
                     # 混音成功，使用混音后的文件
                     output_path = final_path
                 except Exception as mux_err:
-                    logger.warning("音频混合失败，使用无音频版本: %s", mux_err)
+                    logger.warning("[task_id=%s] [混音] 音频混合失败，使用无音频版本: %s", task_id, mux_err)
                     output_path = render_result.output_path
             else:
                 output_path = render_result.output_path
@@ -617,9 +760,15 @@ def generate_video(self, task_id: str) -> dict:
             duration = probe_duration(output_path)
 
             # 5. 上传到 OSS — 失败必须抛异常，不能静默忽略
+            logger.info("[task_id=%s] [OSS上传] 开始上传: size=%d", task_id, file_size)
+            upload_start = time.monotonic()
             file_url = upload_to_oss(output_path, storage_key)
+            upload_elapsed = time.monotonic() - upload_start
             if not file_url:
                 # OSS 未配置或上传失败
+                if gen_task:
+                    gen_task.append_log("OSS上传", "上传失败", level="ERROR")
+                    _flush_logs(task_id, gen_task)
                 raise RuntimeError(
                     f"OSS 上传失败: task_id={task_id}, storage_key={storage_key}, " f"output_path={output_path}"
                 )
@@ -637,13 +786,33 @@ def generate_video(self, task_id: str) -> dict:
                     logger.info(
                         "URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", key
                     )
+                    if gen_task:
+                        gen_task.append_log("OSS上传", "URL校验降级: object_exists确认存在", level="WARN")
                 else:
+                    if gen_task:
+                        gen_task.append_log("OSS上传", "上传后URL不可访问", level="ERROR", file_url=file_url)
+                        _flush_logs(task_id, gen_task)
                     raise RuntimeError(
                         f"OSS 上传后 URL 不可访问且 object_exists 失败: file_url={file_url}, "
                         f"storage_key={storage_key}"
                     )
 
-            logger.info("OSS 上传成功: file_url=%s", file_url)
+            logger.info(
+                "[task_id=%s] [OSS上传] 成功: 耗时=%.1fs, file_url=%s",
+                task_id,
+                upload_elapsed,
+                file_url,
+            )
+
+            if gen_task:
+                gen_task.append_log(
+                    "OSS上传",
+                    f"上传成功, 大小={file_size}, 耗时={upload_elapsed:.1f}s",
+                    file_size=file_size,
+                    duration=round(upload_elapsed, 2),
+                    file_url=file_url,
+                )
+                _flush_logs(task_id, gen_task)
 
             # 6. 创建 GeneratedVideo 记录 + 查重
             dedup_session = SessionLocal()
@@ -665,7 +834,23 @@ def generate_video(self, task_id: str) -> dict:
             # 7. 标记任务为 completed
             _update_task_status(task_id, "mark_completed", result_count=video_count or 1)
 
-            logger.info("视频生成完成: task_id=%s duration=%.2fs file_size=%d", task_id, duration, file_size)
+            # 记录完成日志
+            if gen_task:
+                gen_task.append_log(
+                    "任务完成",
+                    f"视频生成完成: 时长={duration:.2f}s, 大小={file_size}",
+                    duration=round(duration, 2),
+                    file_size=file_size,
+                    video_count=video_count or 1,
+                )
+                _flush_logs(task_id, gen_task)
+
+            logger.info(
+                "[task_id=%s] [任务完成] duration=%.2fs file_size=%d",
+                task_id,
+                duration,
+                file_size,
+            )
 
             return {
                 "status": "completed",
@@ -678,7 +863,27 @@ def generate_video(self, task_id: str) -> dict:
                 "mode": editing_mode.value,
             }
     except Exception as error:
-        logger.error("Video generation failed: %s", error, exc_info=True)
+        logger.error("[task_id=%s] [任务失败] %s", task_id, error, exc_info=True)
+
+        # 记录失败日志
+        try:
+            _session = SessionLocal()
+            try:
+                _repo = SQLAlchemyGenerationTaskRepository(_session)
+                gen_task = _repo.get(task_id)
+                if gen_task:
+                    gen_task.append_log(
+                        "任务失败",
+                        str(error),
+                        level="ERROR",
+                        error_type=type(error).__name__,
+                    )
+                    _flush_logs(task_id, gen_task)
+            finally:
+                _session.close()
+        except Exception:
+            logger.warning("[task_id=%s] 记录失败日志异常", task_id, exc_info=True)
+
         _update_task_status(task_id, "mark_failed", error_message=str(error))
         return {
             "status": "failed",

docs/schema-metadata-snapshot.json

@@ -1549,6 +1549,14 @@
           "type": "JSON",
           "unique": false
         },
+        {
+          "index": false,
+          "name": "logs",
+          "nullable": false,
+          "primary_key": false,
+          "type": "TEXT",
+          "unique": false
+        },
         {
           "index": false,
           "name": "created_at",

packages/adapters/sqlalchemy_impl/generation_task_repository.py

@@ -27,6 +27,7 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
         source_edit_plan_id=model.source_edit_plan_id or "",
         asset_select_mode=model.asset_select_mode or "",
         batch_id=model.batch_id or "",
+        logs=model.logs or "[]",
         created_at=model.created_at,
     )
 
@@ -56,6 +57,7 @@ class SQLAlchemyGenerationTaskRepository:
             source_edit_plan_id=task.source_edit_plan_id or None,
             asset_select_mode=task.asset_select_mode or "",
             batch_id=task.batch_id or "",
+            logs=task.logs,
             created_at=task.created_at,
         )
         self.session.add(model)
@@ -129,5 +131,6 @@ class SQLAlchemyGenerationTaskRepository:
         model.source_edit_plan_id = task.source_edit_plan_id or None
         model.asset_select_mode = task.asset_select_mode or ""
         model.batch_id = task.batch_id or ""
+        model.logs = task.logs
         self.session.commit()
         return task

packages/adapters/sqlalchemy_impl/models.py

@@ -257,6 +257,7 @@ class GenerationTaskModel(Base):
     asset_select_mode = Column(String(20), nullable=False, default="")
     batch_id = Column(String(36), nullable=False, default="", index=True)
     extra_meta = Column("metadata", JSON, nullable=False, default=dict)
+    logs = Column(Text, nullable=False, default="[]", server_default="[]")
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
 
 

packages/domain/generation_task.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import json
 import sys
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
@@ -85,6 +86,7 @@ class GenerationTask:
     created_by_user_id: str = ""
     asset_select_mode: str = ""
     batch_id: str = ""
+    logs: str = "[]"
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 
     @classmethod
@@ -227,6 +229,43 @@ class GenerationTask:
         self.transition_to(GenerationTaskStatus.CANCELLED)
         self.completed_at = datetime.now(timezone.utc)
 
+    # ── 日志辅助 ────────────────────────────────────────────────────────────
+
+    _MAX_LOGS = 200
+
+    def append_log(self, stage: str, message: str, level: str = "INFO", **kwargs) -> None:
+        """追加一条结构化日志到 logs 字段。
+
+        Args:
+            stage: 阶段名称（如 "接收任务"、"下载素材"、"渲染"）
+            message: 日志消息
+            level: 日志级别（INFO / WARN / ERROR）
+            **kwargs: 额外字段（如 asset_id、duration 等）
+        """
+        try:
+            entries = json.loads(self.logs) if self.logs else []
+        except (json.JSONDecodeError, TypeError):
+            entries = []
+        entry = {
+            "ts": datetime.now(timezone.utc).isoformat(),
+            "level": level,
+            "stage": stage,
+            "message": message,
+            **kwargs,
+        }
+        entries.append(entry)
+        # 限制最多保留 _MAX_LOGS 条，防止字段过大
+        if len(entries) > self._MAX_LOGS:
+            entries = entries[-self._MAX_LOGS :]
+        self.logs = json.dumps(entries, ensure_ascii=False)
+
+    def get_logs(self) -> list[dict]:
+        """解析 logs 字段为 list[dict]。"""
+        try:
+            return json.loads(self.logs) if self.logs else []
+        except (json.JSONDecodeError, TypeError):
+            return []
+
     def mark_pending_from_failed(self) -> None:
         """从失败状态重置为待处理（用于重试）。
 

tests/unit/test_generation_logging.py

@@ -0,0 +1,244 @@
+"""
+一键生成链路日志最小集 单元测试
+
+覆盖：
+- GenerationTask.append_log() 正确追加结构化日志
+- GenerationTask.append_log() 超过 200 条时截断
+- GenerationTask.get_logs() 正确解析 JSON
+- GenerationTask.get_logs() 异常 JSON 不抛异常
+- GenerationTaskResponse logs 字段 validator 解析 JSON 字符串
+- GenerationTaskResponse logs 字段 validator 处理非法输入
+- Worker 日志格式 [task_id=xxx] [阶段] 消息
+- _flush_logs 异常不抛出
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+
+from app.schemas.generation_task import GenerationTaskResponse
+
+from packages.domain.generation_task import GenerationTask, GenerationTaskStatus
+
+
+def _make_task(**kwargs) -> GenerationTask:
+    """创建测试用 GenerationTask。"""
+    defaults = {
+        "id": "task-001",
+        "project_id": "proj-001",
+        "asset_library_id": "lib-001",
+        "strategy_id": "one_take",
+        "voice_library_id": "",
+        "template_id": "tpl-001",
+        "asset_ids": ["asset-1", "asset-2"],
+        "title_ids": [],
+        "voice_ids": [],
+        "status": GenerationTaskStatus.PENDING,
+        "progress": 0.0,
+        "result_count": 0,
+        "error_message": "",
+        "created_by_user_id": "user-001",
+        "source_edit_plan_id": "",
+        "asset_select_mode": "all",
+        "batch_id": "",
+    }
+    defaults.update(kwargs)
+    return GenerationTask(**defaults)
+
+
+class TestAppendLog:
+    """GenerationTask.append_log() 单元测试。"""
+
+    def test_append_single_log(self):
+        task = _make_task()
+        task.append_log("接收任务", "任务开始", mode="one_take")
+
+        logs = task.get_logs()
+        assert len(logs) == 1
+        entry = logs[0]
+        assert entry["level"] == "INFO"
+        assert entry["stage"] == "接收任务"
+        assert entry["message"] == "任务开始"
+        assert entry["mode"] == "one_take"
+        assert "ts" in entry
+
+    def test_append_multiple_logs(self):
+        task = _make_task()
+        task.append_log("接收任务", "任务开始")
+        task.append_log("下载素材", "下载完成", count=3)
+        task.append_log("渲染", "渲染完成", duration=12.5)
+
+        logs = task.get_logs()
+        assert len(logs) == 3
+        assert logs[0]["stage"] == "接收任务"
+        assert logs[1]["stage"] == "下载素材"
+        assert logs[1]["count"] == 3
+        assert logs[2]["stage"] == "渲染"
+        assert logs[2]["duration"] == 12.5
+
+    def test_append_log_with_error_level(self):
+        task = _make_task()
+        task.append_log("任务失败", "OSS上传失败", level="ERROR", error_type="RuntimeError")
+
+        logs = task.get_logs()
+        assert len(logs) == 1
+        assert logs[0]["level"] == "ERROR"
+        assert logs[0]["error_type"] == "RuntimeError"
+
+    def test_append_log_truncates_at_200(self):
+        task = _make_task()
+        for i in range(250):
+            task.append_log("阶段", f"消息{i}")
+
+        logs = task.get_logs()
+        assert len(logs) == 200
+        # 保留最后 200 条
+        assert logs[0]["message"] == "消息50"
+        assert logs[-1]["message"] == "消息249"
+
+    def test_append_log_handles_corrupted_json(self):
+        task = _make_task(logs="not-valid-json")
+        task.append_log("接收任务", "任务开始")
+
+        logs = task.get_logs()
+        assert len(logs) == 1
+        assert logs[0]["message"] == "任务开始"
+
+    def test_append_log_handles_empty_string(self):
+        task = _make_task(logs="")
+        task.append_log("接收任务", "任务开始")
+
+        logs = task.get_logs()
+        assert len(logs) == 1
+
+
+class TestGetLogs:
+    """GenerationTask.get_logs() 单元测试。"""
+
+    def test_get_logs_empty(self):
+        task = _make_task()
+        assert task.get_logs() == []
+
+    def test_get_logs_parses_json(self):
+        entries = [{"ts": "2026-01-01T00:00:00Z", "level": "INFO", "stage": "test", "message": "hello"}]
+        task = _make_task(logs=json.dumps(entries, ensure_ascii=False))
+        logs = task.get_logs()
+        assert len(logs) == 1
+        assert logs[0]["message"] == "hello"
+
+    def test_get_logs_handles_invalid_json(self):
+        task = _make_task(logs="{broken")
+        assert task.get_logs() == []
+
+    def test_get_logs_handles_none(self):
+        task = _make_task(logs=None)
+        assert task.get_logs() == []
+
+
+class TestGenerationTaskResponseLogs:
+    """GenerationTaskResponse logs 字段 validator 测试。"""
+
+    def _make_response_data(self, logs_value) -> dict:
+        return {
+            "id": "task-001",
+            "project_id": "proj-001",
+            "asset_library_id": "lib-001",
+            "strategy_id": "one_take",
+            "voice_library_id": "",
+            "template_id": "",
+            "asset_ids": [],
+            "title_ids": [],
+            "voice_ids": [],
+            "source_edit_plan_id": "",
+            "asset_select_mode": "all",
+            "batch_id": "",
+            "status": "completed",
+            "progress": 1.0,
+            "result_count": 1,
+            "error_message": "",
+            "logs": logs_value,
+        }
+
+    def test_logs_json_string_parsed(self):
+        entries = [{"ts": "2026-01-01T00:00:00Z", "level": "INFO", "stage": "test", "message": "ok"}]
+        data = self._make_response_data(json.dumps(entries, ensure_ascii=False))
+        resp = GenerationTaskResponse(**data)
+        assert isinstance(resp.logs, list)
+        assert len(resp.logs) == 1
+        assert resp.logs[0]["message"] == "ok"
+
+    def test_logs_list_passthrough(self):
+        entries = [{"ts": "2026-01-01T00:00:00Z", "level": "INFO", "stage": "test", "message": "ok"}]
+        data = self._make_response_data(entries)
+        resp = GenerationTaskResponse(**data)
+        assert resp.logs == entries
+
+    def test_logs_invalid_json_returns_empty(self):
+        data = self._make_response_data("{broken")
+        resp = GenerationTaskResponse(**data)
+        assert resp.logs == []
+
+    def test_logs_empty_string_returns_empty(self):
+        data = self._make_response_data("")
+        resp = GenerationTaskResponse(**data)
+        assert resp.logs == []
+
+    def test_logs_default_empty(self):
+        data = self._make_response_data("[]")
+        resp = GenerationTaskResponse(**data)
+        assert resp.logs == []
+
+
+class TestWorkerLogFormat:
+    """Worker 日志格式 [task_id=xxx] [阶段] 消息 测试。"""
+
+    def test_log_format_pattern(self):
+        """验证日志格式匹配 [task_id=xxx] [阶段] 消息。"""
+        import re
+
+        task_id = "abc123"
+        stage = "下载素材"
+        message = "完成: 成功=3个, 耗时=1.5s"
+        formatted = f"[task_id={task_id}] [{stage}] {message}"
+
+        pattern = r"^\[task_id=[\w-]+\] \[.+\] .+$"
+        assert re.match(pattern, formatted)
+
+    def test_log_entries_contain_required_fields(self):
+        """验证 append_log 生成的条目包含所有必需字段。"""
+        task = _make_task()
+        task.append_log("OSS上传", "上传成功", file_size=1024000, duration=2.5)
+
+        logs = task.get_logs()
+        entry = logs[0]
+        assert "ts" in entry
+        assert "level" in entry
+        assert "stage" in entry
+        assert "message" in entry
+        assert entry["file_size"] == 1024000
+        assert entry["duration"] == 2.5
+
+
+class TestFlushLogs:
+    """_flush_logs 异常安全测试。"""
+
+    def test_flush_logs_exception_not_raised(self):
+        """_flush_logs 在 DB 异常时不应抛出。"""
+        # 模拟 worker 环境
+        sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+
+        from worker_app.tasks.generation import _flush_logs
+
+        task = _make_task()
+        task.append_log("测试", "消息")
+
+        # Mock SessionLocal 抛异常
+        with patch("worker_app.tasks.generation.SessionLocal", side_effect=RuntimeError("DB error")):
+            # 不应抛出
+            _flush_logs("task-001", task)