alembic/versions/018_add_jobs_table.py

@@ -0,0 +1,54 @@
+"""Phase 8 任务 2.10: Create jobs table for unified async task management
+
+Revision ID: 018
+Revises: 017
+Create Date: 2026-07-01
+
+新增 jobs 表，用于统一管理异步任务（视频合成、渲染等）的生命周期。
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "018"
+down_revision = "017"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "jobs",
+        sa.Column("id", sa.String(32), primary_key=True),
+        sa.Column("project_id", sa.String(32), nullable=False, index=True),
+        sa.Column("job_type", sa.String(30), nullable=False, index=True),
+        sa.Column("status", sa.String(20), nullable=False, server_default="pending", index=True),
+        sa.Column("progress", sa.Float, nullable=False, server_default="0.0"),
+        sa.Column("current_stage", sa.String(200), nullable=False, server_default=""),
+        sa.Column("payload", sa.JSON, nullable=False, server_default="{}"),
+        sa.Column("result", sa.JSON, nullable=False, server_default="{}"),
+        sa.Column("error_message", sa.Text, nullable=False, server_default=""),
+        sa.Column("retry_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("max_retries", sa.Integer, nullable=False, server_default="3"),
+        sa.Column("celery_task_id", sa.String(100), nullable=False, server_default=""),
+        sa.Column("source_id", sa.String(32), nullable=False, server_default="", index=True),
+        sa.Column("created_by_user_id", sa.String(32), nullable=False, server_default="", index=True),
+        sa.Column("started_at", sa.DateTime, nullable=True),
+        sa.Column("completed_at", sa.DateTime, nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime,
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime,
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("jobs")

apps/api/app/api/router.py

@@ -11,6 +11,7 @@ from app.api.routes.edit_templates import router as edit_templates_router
 from app.api.routes.generated_videos import router as generated_videos_router
 from app.api.routes.generation_tasks import router as generation_tasks_router
 from app.api.routes.health import router as health_check_router
+from app.api.routes.jobs import router as jobs_router
 from app.api.routes.ingest_jobs import router as ingest_jobs_router
 from app.api.routes.projects import router as projects_router
 from app.api.routes.recipes import router as recipes_router
@@ -78,6 +79,10 @@ api_router.include_router(
     prefix="/generation",
     tags=["Generation"],
 )
+api_router.include_router(
+    jobs_router,
+    tags=["Job"],
+)
 api_router.include_router(
     generated_videos_router,
     prefix="/generated-videos",

apps/api/app/api/routes/jobs.py

@@ -0,0 +1,334 @@
+"""Job API 路由 — Phase 8 任务 2.10.
+
+提供统一异步任务管理 RESTful 接口：
+- POST   /api/v1/jobs                        创建任务
+- GET    /api/v1/jobs/{job_id}                任务详情
+- GET    /api/v1/projects/{project_id}/jobs   项目任务列表
+- GET    /api/v1/projects/{project_id}/jobs/stats  任务统计
+- PUT    /api/v1/jobs/{job_id}/progress       更新进度
+- POST   /api/v1/jobs/{job_id}/complete       标记完成
+- POST   /api/v1/jobs/{job_id}/fail           标记失败
+- POST   /api/v1/jobs/{job_id}/retry          重试任务
+- POST   /api/v1/jobs/{job_id}/cancel         取消任务
+- POST   /api/v1/jobs/{job_id}/submit         提交执行
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.auth import AuthenticatedUser, get_current_user
+from app.core.celery_app import celery_app
+from app.dependencies import get_db_session, get_job_repository, get_project_repository
+from app.schemas.job import (
+    CompleteJobRequest,
+    CreateJobRequest,
+    FailJobRequest,
+    JobResponse,
+    JobStatisticsResponse,
+    ListJobsResponse,
+    UpdateProgressRequest,
+    job_to_response,
+)
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.orm import Session
+
+from packages.application.jobs import (
+    CancelJobUseCase,
+    CompleteJobCommand,
+    CompleteJobUseCase,
+    CreateJobCommand,
+    CreateJobUseCase,
+    FailJobCommand,
+    FailJobUseCase,
+    GetJobStatisticsUseCase,
+    GetJobUseCase,
+    ListJobsUseCase,
+    RetryJobUseCase,
+    SubmitJobUseCase,
+    UpdateJobProgressCommand,
+    UpdateJobProgressUseCase,
+)
+from packages.domain.job import JobType
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+# 任务类型 → Celery task name 映射
+_JOB_TYPE_TO_CELERY_TASK: dict[str, str] = {
+    JobType.VIDEO_COMPOSE: "worker.compose_video",
+    JobType.RENDER_EDIT_PLAN: "worker.render_edit_plan",
+    JobType.ASSET_INGEST: "worker.ingest_asset",
+    JobType.CLASSIFICATION: "worker.classify_asset",
+    JobType.VOICE_EXTRACTION: "worker.extract_voice",
+    JobType.GENERATION: "worker.generate_video",
+}
+
+
+def _check_project_access(project_id: str, user_id: str, project_repository) -> None:
+    """检查用户是否有项目访问权限。"""
+    project = project_repository.find_by_id(project_id)
+    if project is None:
+        raise HTTPException(status_code=404, detail=f"Project {project_id} not found")
+    if not project.can_access(user_id):
+        raise HTTPException(status_code=403, detail="Access denied to project")
+
+
+# ── 创建任务 ──────────────────────────────────────────────────────────────────
+
+
+@router.post("/jobs", response_model=JobResponse, status_code=status.HTTP_201_CREATED)
+def create_job(
+    request: CreateJobRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    job_repo: Any = Depends(get_job_repository),
+    project_repository: Any = Depends(get_project_repository),
+) -> JobResponse:
+    """创建异步任务。
+
+    创建后任务处于 pending 状态，需要调用 /submit 提交执行。
+    """
+    _check_project_access(request.project_id, authenticated_user.user.id, project_repository)
+
+    # 校验 job_type
+    try:
+        JobType(request.job_type)
+    except ValueError:
+        raise HTTPException(
+            status_code=400,
+            detail=f"不支持的任务类型: {request.job_type}，"
+                   f"可选值: {[t.value for t in JobType]}",
+        )
+
+    use_case = CreateJobUseCase(job_repo)
+    job = use_case.execute(
+        CreateJobCommand(
+            project_id=request.project_id,
+            job_type=request.job_type,
+            payload=request.payload,
+            source_id=request.source_id,
+            created_by_user_id=authenticated_user.user.id,
+            max_retries=request.max_retries,
+        )
+    )
+
+    return job_to_response(job)
+
+
+# ── 提交执行 ──────────────────────────────────────────────────────────────────
+
+
+@router.post("/jobs/{job_id}/submit", response_model=JobResponse)
+def submit_job(
+    job_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    job_repo: Any = Depends(get_job_repository),
+) -> JobResponse:
+    """提交任务执行。
+
+    将任务状态从 pending 切换为 running，并 dispatch Celery 异步任务。
+    """
+    # 权限检查：先获取任务并验证权限，再执行状态变更
+    job = job_repo.get(job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail=f"Job {job_id} not found")
+    if job.created_by_user_id and job.created_by_user_id != authenticated_user.user.id:
+        raise HTTPException(status_code=403, detail="Access denied to this job")
+
+    use_case = SubmitJobUseCase(job_repo)
+
+    try:
+        job = use_case.execute(job_id)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+    # Dispatch Celery 任务
+    celery_task_name = _JOB_TYPE_TO_CELERY_TASK.get(job.job_type.value)
+    if celery_task_name:
+        result = celery_app.send_task(celery_task_name, args=[job.id], kwargs=job.payload)
+        job.celery_task_id = result.id
+        job_repo.update(job)
+        logger.info("已提交 Celery 任务: job_id=%s celery_task_id=%s", job.id, result.id)
+
+    return job_to_response(job)
+
+
+# ── 查询接口 ──────────────────────────────────────────────────────────────────
+
+
+@router.get("/jobs/{job_id}", response_model=JobResponse)
+def get_job(
+    job_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    job_repo: Any = Depends(get_job_repository),
+) -> JobResponse:
+    """获取任务详情。"""
+    use_case = GetJobUseCase(job_repo)
+    job = use_case.execute(job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail=f"Job {job_id} not found")
+    return job_to_response(job)
+
+
+@router.get("/projects/{project_id}/jobs", response_model=ListJobsResponse)
+def list_project_jobs(
+    project_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    job_repo: Any = Depends(get_job_repository),
+    project_repository: Any = Depends(get_project_repository),
+    job_type: str | None = Query(default=None, description="按任务类型过滤"),
+    status_filter: str | None = Query(default=None, alias="status", description="按状态过滤"),
+    limit: int = Query(default=50, ge=1, le=200),
+    offset: int = Query(default=0, ge=0),
+) -> ListJobsResponse:
+    """获取项目下的任务列表。"""
+    _check_project_access(project_id, authenticated_user.user.id, project_repository)
+
+    use_case = ListJobsUseCase(job_repo)
+    jobs = use_case.execute(
+        project_id=project_id,
+        job_type=job_type,
+        status=status_filter,
+        limit=limit,
+        offset=offset,
+    )
+    items = [job_to_response(j) for j in jobs]
+    return ListJobsResponse(items=items, total=len(items))
+
+
+@router.get("/projects/{project_id}/jobs/stats", response_model=JobStatisticsResponse)
+def get_job_statistics(
+    project_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    job_repo: Any = Depends(get_job_repository),
+    project_repository: Any = Depends(get_project_repository),
+) -> JobStatisticsResponse:
+    """获取项目任务统计摘要。"""
+    _check_project_access(project_id, authenticated_user.user.id, project_repository)
+
+    use_case = GetJobStatisticsUseCase(job_repo)
+    stats = use_case.execute(project_id)
+    return JobStatisticsResponse(**stats)
+
+
+# ── 进度更新 ──────────────────────────────────────────────────────────────────
+
+
+@router.put("/jobs/{job_id}/progress", response_model=JobResponse)
+def update_job_progress(
+    job_id: str,
+    request: UpdateProgressRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    job_repo: Any = Depends(get_job_repository),
+) -> JobResponse:
+    """更新任务进度。"""
+    use_case = UpdateJobProgressUseCase(job_repo)
+
+    try:
+        job = use_case.execute(
+            UpdateJobProgressCommand(
+                job_id=job_id,
+                progress=request.progress,
+                current_stage=request.current_stage,
+            )
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+    return job_to_response(job)
+
+
+# ── 完成 / 失败 ────────────────────────────────────────────────────────────────
+
+
+@router.post("/jobs/{job_id}/complete", response_model=JobResponse)
+def complete_job(
+    job_id: str,
+    request: CompleteJobRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    job_repo: Any = Depends(get_job_repository),
+) -> JobResponse:
+    """标记任务完成。"""
+    use_case = CompleteJobUseCase(job_repo)
+
+    try:
+        job = use_case.execute(CompleteJobCommand(job_id=job_id, result=request.result))
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+    return job_to_response(job)
+
+
+@router.post("/jobs/{job_id}/fail", response_model=JobResponse)
+def fail_job(
+    job_id: str,
+    request: FailJobRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    job_repo: Any = Depends(get_job_repository),
+) -> JobResponse:
+    """标记任务失败。"""
+    use_case = FailJobUseCase(job_repo)
+
+    try:
+        job = use_case.execute(FailJobCommand(job_id=job_id, error_message=request.error_message))
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+    return job_to_response(job)
+
+
+# ── 重试 / 取消 ────────────────────────────────────────────────────────────────
+
+
+@router.post("/jobs/{job_id}/retry", response_model=JobResponse)
+def retry_job(
+    job_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    job_repo: Any = Depends(get_job_repository),
+) -> JobResponse:
+    """重试失败任务。
+
+    将任务重置为 pending，retry_count + 1，但不自动 dispatch。
+    需要再次调用 /submit 提交执行。
+    """
+    # 权限检查：先获取任务并验证权限，再执行状态变更
+    job = job_repo.get(job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail=f"Job {job_id} not found")
+    if job.created_by_user_id and job.created_by_user_id != authenticated_user.user.id:
+        raise HTTPException(status_code=403, detail="Access denied to this job")
+
+    use_case = RetryJobUseCase(job_repo)
+
+    try:
+        job = use_case.execute(job_id)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+    return job_to_response(job)
+
+
+@router.post("/jobs/{job_id}/cancel", response_model=JobResponse)
+def cancel_job(
+    job_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    job_repo: Any = Depends(get_job_repository),
+) -> JobResponse:
+    """取消任务。"""
+    # 权限检查：先获取任务并验证权限，再执行状态变更
+    job = job_repo.get(job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail=f"Job {job_id} not found")
+    if job.created_by_user_id and job.created_by_user_id != authenticated_user.user.id:
+        raise HTTPException(status_code=403, detail="Access denied to this job")
+
+    use_case = CancelJobUseCase(job_repo)
+
+    try:
+        job = use_case.execute(job_id)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+    return job_to_response(job)

apps/api/app/dependencies.py

@@ -34,6 +34,7 @@ from packages.adapters.sqlalchemy_impl.generation_task_repository import (
 from packages.adapters.sqlalchemy_impl.ingest_job_repository import (
     SQLAlchemyIngestJobRepository,
 )
+from packages.adapters.sqlalchemy_impl.job_repository import SQLAlchemyJobRepository
 from packages.adapters.sqlalchemy_impl.project_repository import (
     SQLAlchemyProjectRepository,
 )
@@ -52,6 +53,7 @@ from packages.ports.duplication_repository import DuplicationRecordRepository
 from packages.ports.generated_video_repository import GeneratedVideoRepository
 from packages.ports.generation_task_repository import GenerationTaskRepository
 from packages.ports.ingest_job_repository import IngestJobRepository
+from packages.ports.job_repository import JobRepository
 from packages.ports.project_repository import ProjectRepository
 from packages.ports.title_library_repository import TitleLibraryRepository
 from packages.ports.user_repository import UserRepository
@@ -104,6 +106,13 @@ def get_generation_task_repository(
     return SQLAlchemyGenerationTaskRepository(session)
 
 
+def get_job_repository(
+    session: Session = Depends(get_db_session),
+) -> SQLAlchemyJobRepository:
+    """Provide the SQLAlchemy job repository implementation."""
+    return SQLAlchemyJobRepository(session)
+
+
 def get_generated_video_repository(
     session: Session = Depends(get_db_session),
 ) -> SQLAlchemyGeneratedVideoRepository:

apps/api/app/schemas/job.py

@@ -0,0 +1,109 @@
+"""Job API schemas — Phase 8 任务 2.10."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Optional
+
+from pydantic import BaseModel, Field
+
+
+class CreateJobRequest(BaseModel):
+    """创建任务请求体。"""
+
+    project_id: str = Field(..., min_length=1, description="项目 ID")
+    job_type: str = Field(
+        ...,
+        description="任务类型: video_compose / render_edit_plan / asset_ingest / classification / voice_extraction / generation",
+    )
+    payload: dict[str, Any] = Field(default_factory=dict, description="任务输入参数")
+    source_id: str = Field(default="", description="关联的业务实体 ID（如 edit_plan_id）")
+    max_retries: int = Field(default=3, ge=0, le=10, description="最大重试次数")
+
+
+class UpdateProgressRequest(BaseModel):
+    """更新任务进度请求体。"""
+
+    progress: float = Field(..., ge=0.0, le=100.0, description="进度百分比")
+    current_stage: str = Field(default="", description="当前阶段描述")
+
+
+class CompleteJobRequest(BaseModel):
+    """完成任务请求体。"""
+
+    result: dict[str, Any] = Field(default_factory=dict, description="任务结果")
+
+
+class FailJobRequest(BaseModel):
+    """标记任务失败请求体。"""
+
+    error_message: str = Field(..., min_length=1, description="错误信息")
+
+
+class JobResponse(BaseModel):
+    """任务响应体。"""
+
+    id: str
+    project_id: str
+    job_type: str
+    status: str
+    progress: float
+    current_stage: str
+    payload: dict[str, Any]
+    result: dict[str, Any]
+    error_message: str
+    retry_count: int
+    max_retries: int
+    celery_task_id: str
+    source_id: str
+    created_by_user_id: str
+    is_retryable: bool
+    started_at: Optional[datetime] = None
+    completed_at: Optional[datetime] = None
+    created_at: datetime
+    updated_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class ListJobsResponse(BaseModel):
+    """任务列表响应体。"""
+
+    items: list[JobResponse]
+    total: int
+
+
+class JobStatisticsResponse(BaseModel):
+    """任务统计响应体。"""
+
+    project_id: str
+    total: int
+    pending: int
+    running: int
+    success: int
+    failed: int
+
+
+def job_to_response(job) -> JobResponse:
+    """将 Job 领域对象转换为 API 响应。"""
+    return JobResponse(
+        id=job.id,
+        project_id=job.project_id,
+        job_type=job.job_type.value if hasattr(job.job_type, "value") else str(job.job_type),
+        status=job.status.value if hasattr(job.status, "value") else str(job.status),
+        progress=job.progress,
+        current_stage=job.current_stage,
+        payload=job.payload,
+        result=job.result,
+        error_message=job.error_message,
+        retry_count=job.retry_count,
+        max_retries=job.max_retries,
+        celery_task_id=job.celery_task_id,
+        source_id=job.source_id,
+        created_by_user_id=job.created_by_user_id,
+        is_retryable=job.is_retryable,
+        started_at=job.started_at,
+        completed_at=job.completed_at,
+        created_at=job.created_at,
+        updated_at=job.updated_at,
+    )

apps/api/app/services/__init__.py

@@ -3,11 +3,13 @@
 from .auto_clip_service import AutoClipService
 from .edit_plan_service import EditPlanService
 from .edit_template_service import EditTemplateService
+from .job_service import JobService
 from .video_compose_service import VideoComposeService
 
 __all__ = [
     "AutoClipService",
     "EditPlanService",
     "EditTemplateService",
+    "JobService",
     "VideoComposeService",
 ]

apps/api/app/services/job_service.py

@@ -0,0 +1,268 @@
+"""JobService 服务层 — Phase 8 任务 2.10.
+
+将 JobService 与 VideoComposeService 集成，提供视频合成的完整异步工作流：
+1. 创建 Job（记录任务元数据）
+2. 提交执行（dispatch Celery 任务）
+3. Celery 任务中更新进度、处理完成/失败
+
+同时也提供通用的 Job 管理能力，供 ClipPlanService、RenderOrchestrator 等使用。
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from packages.application.jobs import (
+    CancelJobUseCase,
+    CompleteJobCommand,
+    CompleteJobUseCase,
+    CreateJobCommand,
+    CreateJobUseCase,
+    FailJobCommand,
+    FailJobUseCase,
+    GetJobStatisticsUseCase,
+    GetJobUseCase,
+    ListJobsUseCase,
+    RetryJobUseCase,
+    SubmitJobUseCase,
+    UpdateJobProgressCommand,
+    UpdateJobProgressUseCase,
+)
+from packages.domain.job import Job, JobStatus, JobType
+from packages.ports.job_repository import JobRepository
+
+logger = logging.getLogger(__name__)
+
+
+class JobService:
+    """统一异步任务管理服务。
+
+    职责：
+    - 为视频合成等耗时操作提供统一的异步任务管理
+    - 封装 Use Case 的调用，提供简洁的服务接口
+    - 与 VideoComposeService 集成，支持视频合成工作流
+
+    用法::
+
+        job_service = JobService(db)
+        job = job_service.create_compose_job(
+            project_id="xxx",
+            plan_id="yyy",
+            user_id="zzz",
+        )
+        job_service.submit_job(job.id, celery_task_id="celery-xxx")
+    """
+
+    def __init__(self, job_repo: JobRepository):
+        self._job_repo = job_repo
+
+    # ── 创建任务 ──────────────────────────────────────────────────────────
+
+    def create_compose_job(
+        self,
+        project_id: str,
+        plan_id: str,
+        user_id: str,
+        *,
+        max_retries: int = 3,
+    ) -> Job:
+        """创建视频合成任务。
+
+        Args:
+            project_id: 项目 ID
+            plan_id: EditPlan ID
+            user_id: 创建人 ID
+            max_retries: 最大重试次数
+
+        Returns:
+            创建的 Job 实例
+        """
+        use_case = CreateJobUseCase(self._job_repo)
+        return use_case.execute(
+            CreateJobCommand(
+                project_id=project_id,
+                job_type=JobType.VIDEO_COMPOSE,
+                payload={"plan_id": plan_id},
+                source_id=plan_id,
+                created_by_user_id=user_id,
+                max_retries=max_retries,
+            )
+        )
+
+    def create_render_job(
+        self,
+        project_id: str,
+        plan_id: str,
+        user_id: str,
+        *,
+        max_retries: int = 3,
+    ) -> Job:
+        """创建剪辑计划渲染任务。"""
+        use_case = CreateJobUseCase(self._job_repo)
+        return use_case.execute(
+            CreateJobCommand(
+                project_id=project_id,
+                job_type=JobType.RENDER_EDIT_PLAN,
+                payload={"plan_id": plan_id},
+                source_id=plan_id,
+                created_by_user_id=user_id,
+                max_retries=max_retries,
+            )
+        )
+
+    def create_job(
+        self,
+        project_id: str,
+        job_type: JobType | str,
+        *,
+        payload: dict | None = None,
+        source_id: str = "",
+        user_id: str = "",
+        max_retries: int = 3,
+    ) -> Job:
+        """创建通用任务。"""
+        use_case = CreateJobUseCase(self._job_repo)
+        return use_case.execute(
+            CreateJobCommand(
+                project_id=project_id,
+                job_type=job_type,
+                payload=payload or {},
+                source_id=source_id,
+                created_by_user_id=user_id,
+                max_retries=max_retries,
+            )
+        )
+
+    # ── 提交执行 ──────────────────────────────────────────────────────────
+
+    def submit_job(self, job_id: str, celery_task_id: str = "") -> Job:
+        """提交任务执行。"""
+        use_case = SubmitJobUseCase(self._job_repo)
+        return use_case.execute(job_id, celery_task_id)
+
+    # ── 进度更新 ──────────────────────────────────────────────────────────
+
+    def update_progress(self, job_id: str, progress: float, stage: str = "") -> Job:
+        """更新任务进度。"""
+        use_case = UpdateJobProgressUseCase(self._job_repo)
+        return use_case.execute(
+            UpdateJobProgressCommand(
+                job_id=job_id,
+                progress=progress,
+                current_stage=stage,
+            )
+        )
+
+    # ── 完成 / 失败 ────────────────────────────────────────────────────────
+
+    def complete_job(self, job_id: str, result: dict | None = None) -> Job:
+        """标记任务完成。"""
+        use_case = CompleteJobUseCase(self._job_repo)
+        return use_case.execute(CompleteJobCommand(job_id=job_id, result=result or {}))
+
+    def fail_job(self, job_id: str, error_message: str) -> Job:
+        """标记任务失败。"""
+        use_case = FailJobUseCase(self._job_repo)
+        return use_case.execute(FailJobCommand(job_id=job_id, error_message=error_message))
+
+    # ── 重试 / 取消 ────────────────────────────────────────────────────────
+
+    def retry_job(self, job_id: str) -> Job:
+        """重试失败任务。"""
+        use_case = RetryJobUseCase(self._job_repo)
+        return use_case.execute(job_id)
+
+    def cancel_job(self, job_id: str) -> Job:
+        """取消任务。"""
+        use_case = CancelJobUseCase(self._job_repo)
+        return use_case.execute(job_id)
+
+    # ── 查询 ─────────────────────────────────────────────────────────────
+
+    def get_job(self, job_id: str) -> Job | None:
+        """获取任务详情。"""
+        use_case = GetJobUseCase(self._job_repo)
+        return use_case.execute(job_id)
+
+    def list_project_jobs(
+        self,
+        project_id: str,
+        *,
+        job_type: JobType | str | None = None,
+        status: JobStatus | str | None = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> list[Job]:
+        """获取项目下的任务列表。"""
+        use_case = ListJobsUseCase(self._job_repo)
+        return use_case.execute(
+            project_id=project_id,
+            job_type=job_type,
+            status=status,
+            limit=limit,
+            offset=offset,
+        )
+
+    def list_user_jobs(
+        self,
+        user_id: str,
+        *,
+        job_type: JobType | str | None = None,
+        status: JobStatus | str | None = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> list[Job]:
+        """获取用户的任务列表。"""
+        use_case = ListJobsUseCase(self._job_repo)
+        return use_case.execute(
+            user_id=user_id,
+            job_type=job_type,
+            status=status,
+            limit=limit,
+            offset=offset,
+        )
+
+    def get_statistics(self, project_id: str) -> dict[str, Any]:
+        """获取项目任务统计。"""
+        use_case = GetJobStatisticsUseCase(self._job_repo)
+        return use_case.execute(project_id)
+
+    # ── 防重复检查 ────────────────────────────────────────────────────────
+
+    def has_active_job_for_source(self, source_id: str, job_type: JobType | str) -> bool:
+        """检查是否已有活跃任务（防止重复提交）。
+
+        Args:
+            source_id: 关联的业务实体 ID
+            job_type: 任务类型
+
+        Returns:
+            True 如果存在活跃任务
+        """
+        return self._job_repo.find_active_by_source(source_id, job_type) is not None
+
+    # ── 便捷方法：带防重的视频合成提交 ──────────────────────────────────────
+
+    def submit_compose_if_not_exists(
+        self,
+        project_id: str,
+        plan_id: str,
+        user_id: str,
+        celery_task_id: str = "",
+    ) -> tuple[Job, bool]:
+        """创建并提交视频合成任务（防重复）。
+
+        Returns:
+            (job, created): job 实例和是否新创建的标志
+        """
+        if self.has_active_job_for_source(plan_id, JobType.VIDEO_COMPOSE):
+            existing = self._job_repo.find_active_by_source(plan_id, JobType.VIDEO_COMPOSE)
+            logger.info("已存在活跃的视频合成任务: job_id=%s plan_id=%s", existing.id, plan_id)
+            return existing, False
+
+        job = self.create_compose_job(project_id, plan_id, user_id)
+        job = self.submit_job(job.id, celery_task_id)
+        return job, True

apps/worker/worker_app/celery_app.py

@@ -13,5 +13,6 @@ celery_app.conf.imports = (
     "worker_app.tasks.generation",
     "worker_app.tasks.voice_extraction",
     "worker_app.tasks.edit_plan_generation",
+    "worker_app.tasks.compose_video",
     "apps.worker.video_processing.dedup",
 )

apps/worker/worker_app/tasks/__init__.py

@@ -25,6 +25,10 @@ def __getattr__(name: str):
         from .voice_extraction import extract_voice_task
 
         return extract_voice_task
+    elif name == "compose_video":
+        from .compose_video import compose_video
+
+        return compose_video
     elif name == "extract_background_task":
         from .voice_extraction import extract_background_task
 
@@ -34,6 +38,7 @@ def __getattr__(name: str):
 
 __all__ = [
     "classify_asset",
+    "compose_video",
     "generate_video",
     "healthcheck",
     "ingest_asset",

apps/worker/worker_app/tasks/compose_video.py

@@ -0,0 +1,136 @@
+"""视频合成 Celery 任务 — Phase 8 任务 2.10.
+
+使用 JobService 管理任务生命周期，集成 VideoComposeService 执行合成。
+"""
+
+from __future__ import annotations
+
+import logging
+import shutil
+import subprocess
+import tempfile
+from datetime import datetime, timezone
+from pathlib import Path
+
+from celery.utils.log import get_task_logger
+from worker_app.celery_app import celery_app
+from worker_app.db import SessionLocal
+
+logger = get_task_logger(__name__)
+
+
+def _get_job_service():
+    """延迟导入 JobService，避免循环依赖。"""
+    from packages.adapters.sqlalchemy_impl.job_repository import SQLAlchemyJobRepository
+    from apps.api.app.services.job_service import JobService
+
+    db = SessionLocal()
+    repo = SQLAlchemyJobRepository(db)
+    return JobService(repo), db
+
+
+@celery_app.task(
+    name="worker.compose_video",
+    bind=True,
+    max_retries=3,
+    default_retry_delay=60,
+)
+def compose_video(self, job_id: str, **kwargs):
+    """视频合成任务。
+
+    Args:
+        job_id: JobService 中的任务 ID
+        **kwargs: 来自 Job.payload 的额外参数（plan_id, output_path 等）
+    """
+    job_service, db = _get_job_service()
+
+    try:
+        job = job_service.get_job(job_id)
+        if job is None:
+            logger.error("Job not found: %s", job_id)
+            return {"status": "error", "message": f"Job {job_id} not found"}
+
+        plan_id = job.payload.get("plan_id", "")
+        if not plan_id:
+            job_service.fail_job(job_id, "Missing plan_id in job payload")
+            return {"status": "error", "message": "Missing plan_id"}
+
+        # 标记为 running
+        job_service.update_progress(job_id, progress=10.0, current_stage="初始化合成环境")
+
+        # 延迟导入 VideoComposeService
+        from apps.api.app.services.video_compose_service import VideoComposeService
+
+        compose_svc = VideoComposeService(db)
+
+        # 校验合成条件
+        job_service.update_progress(job_id, progress=20.0, current_stage="校验合成条件")
+        validation = compose_svc.validate_compose(plan_id)
+        if not validation.valid:
+            error_msg = "; ".join(validation.errors)
+            job_service.fail_job(job_id, f"合成校验失败: {error_msg}")
+            return {"status": "error", "message": error_msg}
+
+        # 构建合成命令
+        job_service.update_progress(job_id, progress=30.0, current_stage="构建 FFmpeg 命令")
+        output_path = f"/tmp/video_output/{job_id}.mp4"
+        compose_cmd = compose_svc.build_compose_command(plan_id, output_path)
+
+        # 执行 FFmpeg
+        job_service.update_progress(job_id, progress=50.0, current_stage="正在执行视频合成")
+        logger.info("Executing FFmpeg for job %s, plan %s", job_id, plan_id)
+
+        try:
+            subprocess.run(
+                compose_cmd.command,
+                check=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                timeout=3600,
+            )
+        except subprocess.CalledProcessError as e:
+            job_service.fail_job(job_id, f"FFmpeg 执行失败: {e.stderr[:500]}")
+            raise
+
+        # 上传结果
+        job_service.update_progress(job_id, progress=80.0, current_stage="上传合成结果")
+        storage_key = f"rendered/{plan_id}/{job_id}.mp4"
+
+        from worker_app.tasks.edit_plan_generation import _upload_to_oss
+
+        output_url = _upload_to_oss(Path(output_path), storage_key)
+
+        # 更新 Job 状态为完成
+        result_data = {
+            "plan_id": plan_id,
+            "output_path": output_path,
+            "storage_key": storage_key,
+            "output_url": output_url or "",
+            "estimated_duration": compose_cmd.estimated_duration,
+            "clip_count": len(compose_cmd.clip_chains),
+        }
+        job_service.complete_job(job_id, result=result_data)
+
+        logger.info("视频合成完成: job_id=%s, plan_id=%s", job_id, plan_id)
+        return {"status": "completed", "job_id": job_id, "result": result_data}
+
+    except self.retry_exc as exc:
+        logger.warning("视频合成重试中: job_id=%s, exc=%s", job_id, exc)
+        raise
+    except Exception as exc:
+        logger.exception("视频合成异常: job_id=%s", job_id)
+        try:
+            job_service.fail_job(job_id, str(exc)[:500])
+        except Exception:
+            logger.exception("更新 Job 失败状态时出错")
+        raise self.retry(exc=exc, countdown=60)
+    finally:
+        db.close()
+        # 清理临时文件
+        try:
+            output_path = f"/tmp/video_output/{job_id}.mp4"
+            if Path(output_path).exists():
+                Path(output_path).unlink()
+        except Exception:
+            pass

packages/adapters/sqlalchemy_impl/job_repository.py

@@ -0,0 +1,169 @@
+"""JobRepository SQLAlchemy 适配器 — Phase 8 任务 2.10."""
+
+from __future__ import annotations
+
+from sqlalchemy.orm import Session
+
+from packages.adapters.sqlalchemy_impl.models import JobModel
+from packages.domain.job import Job, JobStatus, JobType
+
+
+def _to_domain(model: JobModel) -> Job:
+    """ORM 模型 → 领域实体。"""
+    return Job(
+        id=model.id,
+        project_id=model.project_id,
+        job_type=JobType(model.job_type),
+        status=JobStatus(model.status),
+        progress=float(model.progress or 0.0),
+        current_stage=model.current_stage or "",
+        payload=dict(model.payload or {}),
+        result=dict(model.result or {}),
+        error_message=model.error_message or "",
+        retry_count=int(model.retry_count or 0),
+        max_retries=int(model.max_retries or 3),
+        celery_task_id=model.celery_task_id or "",
+        source_id=model.source_id or "",
+        created_by_user_id=model.created_by_user_id or "",
+        started_at=model.started_at,
+        completed_at=model.completed_at,
+        created_at=model.created_at,
+        updated_at=model.updated_at,
+    )
+
+
+def _to_model(job: Job) -> JobModel:
+    """领域实体 → ORM 模型。"""
+    return JobModel(
+        id=job.id,
+        project_id=job.project_id,
+        job_type=job.job_type.value,
+        status=job.status.value,
+        progress=job.progress,
+        current_stage=job.current_stage,
+        payload=job.payload,
+        result=job.result,
+        error_message=job.error_message,
+        retry_count=job.retry_count,
+        max_retries=job.max_retries,
+        celery_task_id=job.celery_task_id,
+        source_id=job.source_id,
+        created_by_user_id=job.created_by_user_id,
+        started_at=job.started_at,
+        completed_at=job.completed_at,
+        created_at=job.created_at,
+        updated_at=job.updated_at,
+    )
+
+
+class SQLAlchemyJobRepository:
+    """JobRepository 的 SQLAlchemy 实现。"""
+
+    def __init__(self, session: Session):
+        self.session = session
+
+    def create(self, job: Job) -> Job:
+        model = _to_model(job)
+        self.session.add(model)
+        self.session.commit()
+        return job
+
+    def get(self, job_id: str) -> Job | None:
+        model = self.session.query(JobModel).filter(JobModel.id == job_id).first()
+        if model is None:
+            return None
+        return _to_domain(model)
+
+    def update(self, job: Job) -> Job:
+        model = self.session.query(JobModel).filter(JobModel.id == job.id).first()
+        if model is None:
+            raise ValueError(f"Job {job.id} not found")
+        model.project_id = job.project_id
+        model.job_type = job.job_type.value
+        model.status = job.status.value
+        model.progress = job.progress
+        model.current_stage = job.current_stage
+        model.payload = job.payload
+        model.result = job.result
+        model.error_message = job.error_message
+        model.retry_count = job.retry_count
+        model.max_retries = job.max_retries
+        model.celery_task_id = job.celery_task_id
+        model.source_id = job.source_id
+        model.created_by_user_id = job.created_by_user_id
+        model.started_at = job.started_at
+        model.completed_at = job.completed_at
+        model.updated_at = job.updated_at
+        self.session.commit()
+        return job
+
+    def list_by_project(
+        self,
+        project_id: str,
+        *,
+        job_type: JobType | str | None = None,
+        status: JobStatus | str | None = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> list[Job]:
+        query = self.session.query(JobModel).filter(JobModel.project_id == project_id)
+        if job_type is not None:
+            jt = job_type.value if isinstance(job_type, JobType) else job_type
+            query = query.filter(JobModel.job_type == jt)
+        if status is not None:
+            st = status.value if isinstance(status, JobStatus) else status
+            query = query.filter(JobModel.status == st)
+        models = query.order_by(JobModel.created_at.desc()).offset(offset).limit(limit).all()
+        return [_to_domain(m) for m in models]
+
+    def list_by_user(
+        self,
+        user_id: str,
+        *,
+        job_type: JobType | str | None = None,
+        status: JobStatus | str | None = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> list[Job]:
+        query = self.session.query(JobModel).filter(
+            JobModel.created_by_user_id == user_id
+        )
+        if job_type is not None:
+            jt = job_type.value if isinstance(job_type, JobType) else job_type
+            query = query.filter(JobModel.job_type == jt)
+        if status is not None:
+            st = status.value if isinstance(status, JobStatus) else status
+            query = query.filter(JobModel.status == st)
+        models = query.order_by(JobModel.created_at.desc()).offset(offset).limit(limit).all()
+        return [_to_domain(m) for m in models]
+
+    def count_by_project(
+        self,
+        project_id: str,
+        *,
+        status: JobStatus | str | None = None,
+    ) -> int:
+        query = self.session.query(JobModel).filter(JobModel.project_id == project_id)
+        if status is not None:
+            st = status.value if isinstance(status, JobStatus) else status
+            query = query.filter(JobModel.status == st)
+        return query.count()
+
+    def find_active_by_source(
+        self,
+        source_id: str,
+        job_type: JobType | str,
+    ) -> Job | None:
+        jt = job_type.value if isinstance(job_type, JobType) else job_type
+        model = (
+            self.session.query(JobModel)
+            .filter(
+                JobModel.source_id == source_id,
+                JobModel.job_type == jt,
+                JobModel.status.in_(["pending", "running"]),
+            )
+            .first()
+        )
+        if model is None:
+            return None
+        return _to_domain(model)

packages/adapters/sqlalchemy_impl/models.py

@@ -384,3 +384,27 @@ class TemplateCategoryModel(Base):
     user_id = Column(String(36), nullable=False, index=True)
     name = Column(String(100), nullable=False)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
+
+class JobModel(Base):
+    """Phase 8 任务 2.10 — 统一异步任务 ORM 模型。"""
+
+    __tablename__ = "jobs"
+
+    id = Column(String(32), primary_key=True)
+    project_id = Column(String(32), nullable=False, index=True)
+    job_type = Column(String(30), nullable=False, index=True)
+    status = Column(String(20), nullable=False, default="pending", index=True)
+    progress = Column(Float, nullable=False, default=0.0)
+    current_stage = Column(String(200), nullable=False, default="")
+    payload = Column(JSON, nullable=False, default=dict)
+    result = Column(JSON, nullable=False, default=dict)
+    error_message = Column(Text, nullable=False, default="")
+    retry_count = Column(Integer, nullable=False, default=0)
+    max_retries = Column(Integer, nullable=False, default=3)
+    celery_task_id = Column(String(100), nullable=False, default="")
+    source_id = Column(String(32), nullable=False, default="", index=True)
+    created_by_user_id = Column(String(32), nullable=False, default="", index=True)
+    started_at = Column(DateTime, nullable=True)
+    completed_at = Column(DateTime, nullable=True)
+    created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
+    updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))

packages/application/__init__.py

@@ -30,33 +30,63 @@ from .generation_tasks import (
     GetGenerationTaskUseCase,
 )
 from .ingest_jobs import SubmitIngestJobCommand, SubmitIngestJobUseCase
+from .jobs import (
+    CancelJobUseCase,
+    CompleteJobCommand,
+    CompleteJobUseCase,
+    CreateJobCommand,
+    CreateJobUseCase,
+    FailJobCommand,
+    FailJobUseCase,
+    GetJobStatisticsUseCase,
+    GetJobUseCase,
+    ListJobsUseCase,
+    RetryJobUseCase,
+    SubmitJobUseCase,
+    UpdateJobProgressCommand,
+    UpdateJobProgressUseCase,
+)
 from .projects import CreateProjectCommand, CreateProjectUseCase, GetProjectUseCase, ListProjectsUseCase
 
 __all__ = [
+    "CancelJobUseCase",
+    "CompleteJobCommand",
+    "CompleteJobUseCase",
     "CreateAssetCommand",
     "CreateAssetLibraryCommand",
     "CreateAssetLibraryUseCase",
     "CreateAssetUseCase",
     "CreateGenerationTaskCommand",
     "CreateGenerationTaskUseCase",
+    "CreateJobCommand",
+    "CreateJobUseCase",
     "CreateProjectCommand",
     "CreateProjectUseCase",
     "DeleteDuplicationRecordUseCase",
+    "FailJobCommand",
+    "FailJobUseCase",
     "GetDuplicationDetailUseCase",
     "GetGeneratedVideoDownloadUrlUseCase",
     "GetGeneratedVideoUseCase",
+    "GetJobStatisticsUseCase",
+    "GetJobUseCase",
     "GetProjectUseCase",
     "ListAssetLibrariesUseCase",
     "ListAssetsUseCase",
     "ListDuplicationRecordsUseCase",
     "ListGeneratedVideosByTaskUseCase",
     "ListGeneratedVideosUseCase",
+    "ListJobsUseCase",
     "ListProjectsUseCase",
     "RetryDuplicationUseCase",
+    "RetryJobUseCase",
     "SubmitClassificationJobCommand",
     "SubmitClassificationJobUseCase",
     "SubmitIngestJobCommand",
     "SubmitIngestJobUseCase",
+    "SubmitJobUseCase",
+    "UpdateJobProgressCommand",
+    "UpdateJobProgressUseCase",
     "UploadForDuplicationCommand",
     "UploadForDuplicationUseCase",
 ]

packages/application/jobs.py

@@ -0,0 +1,257 @@
+"""JobService 应用层用例 — Phase 8 任务 2.10.
+
+提供任务创建、提交、状态管理、进度追踪、失败重试、查询等核心能力。
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+from uuid import uuid4
+
+from packages.domain.job import Job, JobStatus, JobType
+from packages.ports.job_repository import JobRepository
+
+logger = logging.getLogger(__name__)
+
+
+# ── 命令对象 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass(slots=True)
+class CreateJobCommand:
+    """创建任务的命令。"""
+
+    project_id: str
+    job_type: JobType | str
+    payload: dict = field(default_factory=dict)
+    source_id: str = ""
+    created_by_user_id: str = ""
+    max_retries: int = 3
+
+
+@dataclass(slots=True)
+class UpdateJobProgressCommand:
+    """更新任务进度的命令。"""
+
+    job_id: str
+    progress: float
+    current_stage: str = ""
+
+
+@dataclass(slots=True)
+class CompleteJobCommand:
+    """完成任务的命令。"""
+
+    job_id: str
+    result: dict = field(default_factory=dict)
+
+
+@dataclass(slots=True)
+class FailJobCommand:
+    """标记任务失败的命令。"""
+
+    job_id: str
+    error_message: str
+
+
+# ── 用例 ─────────────────────────────────────────────────────────────────────
+
+
+class CreateJobUseCase:
+    """创建异步任务。
+
+    创建后状态为 pending，不自动提交到 Celery（由调用方决定何时 dispatch）。
+    """
+
+    def __init__(self, job_repo: JobRepository):
+        self._job_repo = job_repo
+
+    def execute(self, command: CreateJobCommand) -> Job:
+        job = Job.create(
+            project_id=command.project_id,
+            job_type=command.job_type,
+            payload=command.payload,
+            source_id=command.source_id,
+            created_by_user_id=command.created_by_user_id,
+            max_retries=command.max_retries,
+        )
+        logger.info(
+            "创建任务: job_id=%s type=%s project=%s",
+            job.id, job.job_type.value, job.project_id,
+        )
+        return self._job_repo.create(job)
+
+
+class SubmitJobUseCase:
+    """提交任务执行：将任务状态切换为 running 并返回 Celery 任务 ID 占位。
+
+    实际的 Celery dispatch 由上层（路由层 / 服务层）完成，本用例仅做状态流转。
+    """
+
+    def __init__(self, job_repo: JobRepository):
+        self._job_repo = job_repo
+
+    def execute(self, job_id: str, celery_task_id: str = "") -> Job:
+        job = self._job_repo.get(job_id)
+        if job is None:
+            raise ValueError(f"任务不存在: {job_id}")
+        if job.status != JobStatus.PENDING:
+            raise ValueError(f"只有 pending 状态的任务可以提交，当前状态: {job.status.value}")
+
+        job.mark_running(stage="已提交，等待执行")
+        job.celery_task_id = celery_task_id
+        return self._job_repo.update(job)
+
+
+class UpdateJobProgressUseCase:
+    """更新任务进度。"""
+
+    def __init__(self, job_repo: JobRepository):
+        self._job_repo = job_repo
+
+    def execute(self, command: UpdateJobProgressCommand) -> Job:
+        job = self._job_repo.get(command.job_id)
+        if job is None:
+            raise ValueError(f"任务不存在: {command.job_id}")
+        if job.status != JobStatus.RUNNING:
+            raise ValueError(f"只有 running 状态的任务可以更新进度，当前状态: {job.status.value}")
+
+        job.update_progress(command.progress, command.current_stage)
+        return self._job_repo.update(job)
+
+
+class CompleteJobUseCase:
+    """标记任务成功完成。"""
+
+    def __init__(self, job_repo: JobRepository):
+        self._job_repo = job_repo
+
+    def execute(self, command: CompleteJobCommand) -> Job:
+        job = self._job_repo.get(command.job_id)
+        if job is None:
+            raise ValueError(f"任务不存在: {command.job_id}")
+        if job.status not in (JobStatus.RUNNING, JobStatus.PENDING):
+            raise ValueError(f"只有 running/pending 状态的任务可以标记完成，当前状态: {job.status.value}")
+
+        job.mark_success(result=command.result)
+        logger.info("任务完成: job_id=%s", job.id)
+        return self._job_repo.update(job)
+
+
+class FailJobUseCase:
+    """标记任务失败。"""
+
+    def __init__(self, job_repo: JobRepository):
+        self._job_repo = job_repo
+
+    def execute(self, command: FailJobCommand) -> Job:
+        job = self._job_repo.get(command.job_id)
+        if job is None:
+            raise ValueError(f"任务不存在: {command.job_id}")
+
+        job.mark_failed(command.error_message)
+        logger.warning("任务失败: job_id=%s error=%s", job.id, command.error_message)
+        return self._job_repo.update(job)
+
+
+class RetryJobUseCase:
+    """重试失败任务。
+
+    将 failed 任务重置为 pending，retry_count + 1。
+    实际的 Celery 重新 dispatch 由调用方完成。
+    """
+
+    def __init__(self, job_repo: JobRepository):
+        self._job_repo = job_repo
+
+    def execute(self, job_id: str) -> Job:
+        job = self._job_repo.get(job_id)
+        if job is None:
+            raise ValueError(f"任务不存在: {job_id}")
+
+        job.prepare_retry()
+        logger.info("任务重试: job_id=%s retry_count=%d", job.id, job.retry_count)
+        return self._job_repo.update(job)
+
+
+class CancelJobUseCase:
+    """取消任务。"""
+
+    def __init__(self, job_repo: JobRepository):
+        self._job_repo = job_repo
+
+    def execute(self, job_id: str) -> Job:
+        job = self._job_repo.get(job_id)
+        if job is None:
+            raise ValueError(f"任务不存在: {job_id}")
+        if job.is_terminal:
+            raise ValueError(f"任务已处于终态，无法取消: {job.status.value}")
+
+        job.mark_cancelled()
+        logger.info("任务取消: job_id=%s", job.id)
+        return self._job_repo.update(job)
+
+
+class GetJobUseCase:
+    """获取单个任务详情。"""
+
+    def __init__(self, job_repo: JobRepository):
+        self._job_repo = job_repo
+
+    def execute(self, job_id: str) -> Job | None:
+        return self._job_repo.get(job_id)
+
+
+class ListJobsUseCase:
+    """列出任务（支持按项目 / 用户 / 类型 / 状态过滤）。"""
+
+    def __init__(self, job_repo: JobRepository):
+        self._job_repo = job_repo
+
+    def execute(
+        self,
+        *,
+        project_id: str | None = None,
+        user_id: str | None = None,
+        job_type: JobType | str | None = None,
+        status: JobStatus | str | None = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> list[Job]:
+        if project_id:
+            return self._job_repo.list_by_project(
+                project_id, job_type=job_type, status=status,
+                limit=limit, offset=offset,
+            )
+        if user_id:
+            return self._job_repo.list_by_user(
+                user_id, job_type=job_type, status=status,
+                limit=limit, offset=offset,
+            )
+        raise ValueError("必须指定 project_id 或 user_id")
+
+
+class GetJobStatisticsUseCase:
+    """获取项目的任务统计摘要。"""
+
+    def __init__(self, job_repo: JobRepository):
+        self._job_repo = job_repo
+
+    def execute(self, project_id: str) -> dict[str, Any]:
+        total = self._job_repo.count_by_project(project_id)
+        pending = self._job_repo.count_by_project(project_id, status=JobStatus.PENDING)
+        running = self._job_repo.count_by_project(project_id, status=JobStatus.RUNNING)
+        success = self._job_repo.count_by_project(project_id, status=JobStatus.SUCCESS)
+        failed = self._job_repo.count_by_project(project_id, status=JobStatus.FAILED)
+
+        return {
+            "project_id": project_id,
+            "total": total,
+            "pending": pending,
+            "running": running,
+            "success": success,
+            "failed": failed,
+        }

packages/domain/__init__.py

@@ -23,6 +23,7 @@ from .entities import (
 )
 from .generated_video import GeneratedVideo
 from .generation_task import GenerationTask, GenerationTaskStatus
+from .job import Job, JobStatus, JobType
 from .template_clip_config import ClipType, TemplateClipConfig, TransitionEffect
 from .title_library import TitleLibraryItem
 from .voice_library import VoiceLibraryItem
@@ -51,6 +52,9 @@ __all__ = [
     "GenerationTaskStatus",
     "IngestJob",
     "IngestJobStatus",
+    "Job",
+    "JobStatus",
+    "JobType",
     "Project",
     "TemplateClipConfig",
     "TransitionEffect",

packages/domain/job.py

@@ -0,0 +1,289 @@
+"""Job 领域模型 — Phase 8 任务 2.10.
+
+统一的异步任务抽象，为视频合成等耗时操作提供异步任务管理能力。
+
+状态机：
+  pending → running → success
+                  ↘ failed → pending (重试)
+       ↘ cancelled
+"""
+
+from __future__ import annotations
+
+import sys
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+
+if sys.version_info >= (3, 11):
+    from enum import StrEnum
+else:
+    from enum import Enum
+
+    class StrEnum(str, Enum):
+        pass
+
+
+from uuid import uuid4
+
+
+class JobType(StrEnum):
+    """任务类型枚举。"""
+
+    VIDEO_COMPOSE = "video_compose"
+    """视频合成（VideoComposeService）"""
+
+    RENDER_EDIT_PLAN = "render_edit_plan"
+    """剪辑计划渲染"""
+
+    ASSET_INGEST = "asset_ingest"
+    """素材导入"""
+
+    CLASSIFICATION = "classification"
+    """素材分类"""
+
+    VOICE_EXTRACTION = "voice_extraction"
+    """语音提取"""
+
+    GENERATION = "generation"
+    """通用生成"""
+
+
+class JobStatus(StrEnum):
+    """任务状态枚举。"""
+
+    PENDING = "pending"
+    RUNNING = "running"
+    SUCCESS = "success"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+# 终态集合
+TERMINAL_STATUSES = frozenset({JobStatus.SUCCESS, JobStatus.FAILED, JobStatus.CANCELLED})
+
+# 合法状态转换
+_VALID_TRANSITIONS: dict[JobStatus, set[JobStatus]] = {
+    JobStatus.PENDING: {JobStatus.RUNNING, JobStatus.SUCCESS, JobStatus.CANCELLED},
+    JobStatus.RUNNING: {JobStatus.SUCCESS, JobStatus.FAILED, JobStatus.CANCELLED},
+    JobStatus.FAILED: {JobStatus.PENDING},  # 重试回到 pending
+}
+
+
+@dataclass(slots=True)
+class Job:
+    """统一异步任务实体。
+
+    Attributes:
+        id: 任务唯一标识
+        project_id: 所属项目
+        job_type: 任务类型
+        status: 当前状态
+        progress: 进度百分比 (0.0 ~ 100.0)
+        current_stage: 当前阶段描述（人类可读）
+        payload: 任务输入参数（JSON 序列化）
+        result: 任务结果（JSON 序列化）
+        error_message: 错误信息
+        retry_count: 已重试次数
+        max_retries: 最大重试次数
+        celery_task_id: Celery 异步任务 ID
+        source_id: 关联的业务实体 ID（如 edit_plan_id, generation_task_id）
+        created_by_user_id: 创建人
+        started_at: 开始执行时间
+        completed_at: 完成时间
+        created_at: 创建时间
+        updated_at: 最后更新时间
+    """
+
+    id: str
+    project_id: str
+    job_type: JobType
+    status: JobStatus = JobStatus.PENDING
+    progress: float = 0.0
+    current_stage: str = ""
+    payload: dict = field(default_factory=dict)
+    result: dict = field(default_factory=dict)
+    error_message: str = ""
+    retry_count: int = 0
+    max_retries: int = 3
+    celery_task_id: str = ""
+    source_id: str = ""
+    created_by_user_id: str = ""
+    started_at: datetime | None = None
+    completed_at: datetime | None = None
+    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+
+    @classmethod
+    def create(
+        cls,
+        project_id: str,
+        job_type: JobType | str,
+        *,
+        payload: dict | None = None,
+        source_id: str = "",
+        created_by_user_id: str = "",
+        max_retries: int = 3,
+    ) -> Job:
+        """创建新任务。
+
+        Args:
+            project_id: 项目 ID
+            job_type: 任务类型
+            payload: 任务输入参数
+            source_id: 关联业务实体 ID
+            created_by_user_id: 创建人 ID
+            max_retries: 最大重试次数
+
+        Returns:
+            新建的 Job 实例
+
+        Raises:
+            ValueError: 参数校验失败
+        """
+        if not project_id.strip():
+            raise ValueError("project_id 不能为空")
+
+        # 兼容字符串类型
+        if isinstance(job_type, str):
+            try:
+                job_type = JobType(job_type)
+            except ValueError:
+                raise ValueError(f"不支持的任务类型: {job_type}")
+
+        return cls(
+            id=uuid4().hex,
+            project_id=project_id.strip(),
+            job_type=job_type,
+            payload=payload or {},
+            source_id=source_id.strip(),
+            created_by_user_id=created_by_user_id.strip(),
+            max_retries=max_retries,
+        )
+
+    @property
+    def is_terminal(self) -> bool:
+        """是否处于终态。"""
+        return self.status in TERMINAL_STATUSES
+
+    @property
+    def is_retryable(self) -> bool:
+        """是否可重试（失败且未超过重试上限）。"""
+        return self.status == JobStatus.FAILED and self.retry_count < self.max_retries
+
+    def transition_to(self, new_status: JobStatus | str) -> None:
+        """执行状态转换。
+
+        Args:
+            new_status: 目标状态
+
+        Raises:
+            ValueError: 非法状态转换
+        """
+        if isinstance(new_status, str):
+            try:
+                new_status = JobStatus(new_status)
+            except ValueError:
+                raise ValueError(f"无效状态: {new_status}")
+
+        allowed = _VALID_TRANSITIONS.get(self.status, set())
+        if new_status not in allowed:
+            raise ValueError(
+                f"非法状态转换: {self.status.value} → {new_status.value}，"
+                f"允许: {{{', '.join(s.value for s in allowed)}}}"
+            )
+
+        now = datetime.now(timezone.utc)
+        self.status = new_status
+        self.updated_at = now
+
+        # 自动设置时间戳
+        if new_status == JobStatus.RUNNING and self.started_at is None:
+            self.started_at = now
+        elif new_status in (JobStatus.SUCCESS, JobStatus.FAILED):
+            self.completed_at = now
+
+    def mark_running(self, stage: str = "") -> None:
+        """标记为执行中。"""
+        self.transition_to(JobStatus.RUNNING)
+        if stage:
+            self.current_stage = stage
+
+    def mark_success(self, result: dict | None = None) -> None:
+        """标记为成功。"""
+        self.transition_to(JobStatus.SUCCESS)
+        self.progress = 100.0
+        self.current_stage = "完成"
+        if result:
+            self.result = result
+
+    def mark_failed(self, error_message: str) -> None:
+        """标记为失败。"""
+        self.transition_to(JobStatus.FAILED)
+        self.error_message = error_message
+        self.current_stage = "失败"
+
+    def mark_cancelled(self) -> None:
+        """标记为取消。"""
+        self.transition_to(JobStatus.CANCELLED)
+        self.current_stage = "已取消"
+
+    def update_progress(self, progress: float, stage: str = "") -> None:
+        """更新进度。
+
+        Args:
+            progress: 进度百分比 (0.0 ~ 100.0)
+            stage: 当前阶段描述
+
+        Raises:
+            ValueError: 进度值非法
+        """
+        if not 0.0 <= progress <= 100.0:
+            raise ValueError(f"进度必须在 0~100 之间，当前: {progress}")
+        self.progress = progress
+        if stage:
+            self.current_stage = stage
+        self.updated_at = datetime.now(timezone.utc)
+
+    def prepare_retry(self) -> None:
+        """准备重试：重置状态为 pending。
+
+        Raises:
+            ValueError: 不可重试
+        """
+        if not self.is_retryable:
+            raise ValueError(
+                f"任务不可重试: status={self.status.value}, "
+                f"retry_count={self.retry_count}, max_retries={self.max_retries}"
+            )
+        self.retry_count += 1
+        self.transition_to(JobStatus.PENDING)
+        self.progress = 0.0
+        self.current_stage = f"第 {self.retry_count} 次重试"
+        self.error_message = ""
+        self.started_at = None
+        self.completed_at = None
+        self.celery_task_id = ""
+
+    def to_dict(self) -> dict:
+        """序列化为字典。"""
+        return {
+            "id": self.id,
+            "project_id": self.project_id,
+            "job_type": self.job_type.value,
+            "status": self.status.value,
+            "progress": self.progress,
+            "current_stage": self.current_stage,
+            "payload": self.payload,
+            "result": self.result,
+            "error_message": self.error_message,
+            "retry_count": self.retry_count,
+            "max_retries": self.max_retries,
+            "celery_task_id": self.celery_task_id,
+            "source_id": self.source_id,
+            "created_by_user_id": self.created_by_user_id,
+            "is_retryable": self.is_retryable,
+            "started_at": self.started_at.isoformat() if self.started_at else None,
+            "completed_at": self.completed_at.isoformat() if self.completed_at else None,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
+        }

packages/ports/job_repository.py

@@ -0,0 +1,64 @@
+"""JobRepository 端口接口 — Phase 8 任务 2.10."""
+
+from __future__ import annotations
+
+from typing import Protocol
+
+from packages.domain.job import Job, JobStatus, JobType
+
+
+class JobRepository(Protocol):
+    """异步任务仓储接口。"""
+
+    def create(self, job: Job) -> Job:
+        """持久化一个新任务。"""
+        ...
+
+    def get(self, job_id: str) -> Job | None:
+        """根据 ID 获取任务。"""
+        ...
+
+    def update(self, job: Job) -> Job:
+        """更新任务状态。"""
+        ...
+
+    def list_by_project(
+        self,
+        project_id: str,
+        *,
+        job_type: JobType | str | None = None,
+        status: JobStatus | str | None = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> list[Job]:
+        """按项目列出任务，支持类型和状态过滤。"""
+        ...
+
+    def list_by_user(
+        self,
+        user_id: str,
+        *,
+        job_type: JobType | str | None = None,
+        status: JobStatus | str | None = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> list[Job]:
+        """按创建人列出任务。"""
+        ...
+
+    def count_by_project(
+        self,
+        project_id: str,
+        *,
+        status: JobStatus | str | None = None,
+    ) -> int:
+        """按项目统计任务数量。"""
+        ...
+
+    def find_active_by_source(
+        self,
+        source_id: str,
+        job_type: JobType | str,
+    ) -> Job | None:
+        """查找关联同一业务实体的活跃任务（pending/running）。"""
+        ...

tests/unit/test_job_service.py

@@ -0,0 +1,578 @@
+"""Unit tests for JobService — Phase 8 任务 2.10.
+
+测试 Job 领域模型、Use Cases 和 Service 层。
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from packages.domain.job import Job, JobStatus, JobType
+from packages.application.jobs import (
+    CancelJobUseCase,
+    CompleteJobCommand,
+    CompleteJobUseCase,
+    CreateJobCommand,
+    CreateJobUseCase,
+    FailJobCommand,
+    FailJobUseCase,
+    GetJobStatisticsUseCase,
+    GetJobUseCase,
+    ListJobsUseCase,
+    RetryJobUseCase,
+    SubmitJobUseCase,
+    UpdateJobProgressCommand,
+    UpdateJobProgressUseCase,
+)
+
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+
+class FakeJobRepo:
+    """内存中的 JobRepository 测试替身。"""
+
+    def __init__(self):
+        self._store: dict[str, Job] = {}
+
+    def create(self, job: Job) -> Job:
+        self._store[job.id] = job
+        return job
+
+    def get(self, job_id: str) -> Job | None:
+        return self._store.get(job_id)
+
+    def update(self, job: Job) -> Job:
+        if job.id not in self._store:
+            raise ValueError(f"Job {job.id} not found")
+        self._store[job.id] = job
+        return job
+
+    def list_by_project(self, project_id, *, job_type=None, status=None, limit=50, offset=0):
+        results = [
+            j for j in self._store.values()
+            if j.project_id == project_id
+            and (job_type is None or j.job_type == job_type or j.job_type == JobType(job_type))
+            and (status is None or j.status == status or j.status == JobStatus(status))
+        ]
+        return results[offset : offset + limit]
+
+    def list_by_user(self, user_id, *, job_type=None, status=None, limit=50, offset=0):
+        results = [
+            j for j in self._store.values()
+            if j.created_by_user_id == user_id
+            and (job_type is None or j.job_type == job_type or j.job_type == JobType(job_type))
+            and (status is None or j.status == status or j.status == JobStatus(status))
+        ]
+        return results[offset : offset + limit]
+
+    def count_by_project(self, project_id, *, status=None):
+        return len([
+            j for j in self._store.values()
+            if j.project_id == project_id
+            and (status is None or j.status == status or j.status == JobStatus(status))
+        ])
+
+    def find_active_by_source(self, source_id, job_type):
+        jt = job_type.value if isinstance(job_type, JobType) else job_type
+        for j in self._store.values():
+            if j.source_id == source_id and j.job_type.value == jt and j.status in (JobStatus.PENDING, JobStatus.RUNNING):
+                return j
+        return None
+
+
+@pytest.fixture
+def repo():
+    return FakeJobRepo()
+
+
+# ── Job 领域模型测试 ──────────────────────────────────────────────────────────
+
+
+class TestJobDomainModel:
+    def test_create_job(self):
+        job = Job.create(
+            project_id="proj-1",
+            job_type=JobType.VIDEO_COMPOSE,
+            payload={"plan_id": "plan-1"},
+            source_id="plan-1",
+            created_by_user_id="user-1",
+        )
+        assert job.project_id == "proj-1"
+        assert job.job_type == JobType.VIDEO_COMPOSE
+        assert job.status == JobStatus.PENDING
+        assert job.progress == 0.0
+        assert job.retry_count == 0
+        assert job.max_retries == 3
+        assert job.payload == {"plan_id": "plan-1"}
+
+    def test_create_job_with_string_type(self):
+        job = Job.create(project_id="proj-1", job_type="render_edit_plan")
+        assert job.job_type == JobType.RENDER_EDIT_PLAN
+
+    def test_create_job_invalid_type_raises(self):
+        with pytest.raises(ValueError, match="不支持的任务类型"):
+            Job.create(project_id="proj-1", job_type="invalid_type")
+
+    def test_create_job_empty_project_id_raises(self):
+        with pytest.raises(ValueError, match="project_id 不能为空"):
+            Job.create(project_id="  ", job_type=JobType.VIDEO_COMPOSE)
+
+    def test_is_terminal(self):
+        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
+        assert not job.is_terminal
+
+        job.status = JobStatus.SUCCESS
+        assert job.is_terminal
+
+        job.status = JobStatus.FAILED
+        assert job.is_terminal
+
+        job.status = JobStatus.CANCELLED
+        assert job.is_terminal
+
+    def test_is_retryable(self):
+        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, max_retries=2)
+        job.status = JobStatus.FAILED
+        assert job.is_retryable
+
+        job.retry_count = 2
+        assert not job.is_retryable
+
+    def test_mark_running(self):
+        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_running("初始化中")
+        assert job.status == JobStatus.RUNNING
+        assert job.current_stage == "初始化中"
+        assert job.started_at is not None
+
+    def test_mark_success(self):
+        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_running()
+        job.mark_success(result={"output_url": "https://example.com/video.mp4"})
+        assert job.status == JobStatus.SUCCESS
+        assert job.progress == 100.0
+        assert job.current_stage == "完成"
+        assert job.result == {"output_url": "https://example.com/video.mp4"}
+        assert job.completed_at is not None
+
+    def test_mark_failed(self):
+        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_running()
+        job.mark_failed("FFmpeg 执行失败")
+        assert job.status == JobStatus.FAILED
+        assert job.error_message == "FFmpeg 执行失败"
+        assert job.completed_at is not None
+
+    def test_mark_cancelled(self):
+        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_cancelled()
+        assert job.status == JobStatus.CANCELLED
+        assert job.current_stage == "已取消"
+
+    def test_update_progress(self):
+        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
+        job.update_progress(50.0, "合成中")
+        assert job.progress == 50.0
+        assert job.current_stage == "合成中"
+
+    def test_update_progress_invalid_value_raises(self):
+        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
+        with pytest.raises(ValueError, match="进度必须在 0~100 之间"):
+            job.update_progress(150.0)
+
+    def test_prepare_retry(self):
+        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, max_retries=3)
+        job.mark_running()
+        job.mark_failed("临时错误")
+        job.prepare_retry()
+        assert job.status == JobStatus.PENDING
+        assert job.retry_count == 1
+        assert job.progress == 0.0
+        assert job.error_message == ""
+        assert "第 1 次重试" in job.current_stage
+
+    def test_prepare_retry_exceeded_raises(self):
+        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, max_retries=1)
+        job.mark_running()
+        job.mark_failed("错误")
+        job.prepare_retry()  # retry_count = 1, max_retries = 1 → not retryable
+        with pytest.raises(ValueError, match="任务不可重试"):
+            job.prepare_retry()
+
+    def test_invalid_transition_raises(self):
+        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_running()
+        job.mark_success()
+        with pytest.raises(ValueError, match="非法状态转换"):
+            job.transition_to(JobStatus.RUNNING)  # success → running 不合法（终态不可转换）
+
+    def test_to_dict(self):
+        job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
+        d = job.to_dict()
+        assert d["project_id"] == "proj-1"
+        assert d["job_type"] == "video_compose"
+        assert d["status"] == "pending"
+        assert "is_retryable" in d
+
+
+# ── Use Case 测试 ─────────────────────────────────────────────────────────────
+
+
+class TestCreateJobUseCase:
+    def test_execute(self, repo):
+        use_case = CreateJobUseCase(repo)
+        cmd = CreateJobCommand(
+            project_id="proj-1",
+            job_type=JobType.VIDEO_COMPOSE,
+            payload={"plan_id": "plan-1"},
+            source_id="plan-1",
+            created_by_user_id="user-1",
+        )
+        job = use_case.execute(cmd)
+        assert job.id in repo._store
+        assert job.status == JobStatus.PENDING
+
+    def test_execute_with_string_type(self, repo):
+        use_case = CreateJobUseCase(repo)
+        cmd = CreateJobCommand(
+            project_id="proj-1",
+            job_type="video_compose",
+        )
+        job = use_case.execute(cmd)
+        assert job.job_type == JobType.VIDEO_COMPOSE
+
+
+class TestSubmitJobUseCase:
+    def test_submit_pending_job(self, repo):
+        # 先创建
+        create_uc = CreateJobUseCase(repo)
+        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+
+        # 提交
+        submit_uc = SubmitJobUseCase(repo)
+        submitted = submit_uc.execute(job.id, celery_task_id="celery-abc")
+        assert submitted.status == JobStatus.RUNNING
+        assert submitted.celery_task_id == "celery-abc"
+        assert submitted.started_at is not None
+
+    def test_submit_non_pending_raises(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        job.mark_running()
+        repo.update(job)
+
+        submit_uc = SubmitJobUseCase(repo)
+        with pytest.raises(ValueError, match="只有 pending 状态"):
+            submit_uc.execute(job.id)
+
+    def test_submit_nonexistent_raises(self, repo):
+        submit_uc = SubmitJobUseCase(repo)
+        with pytest.raises(ValueError, match="任务不存在"):
+            submit_uc.execute("nonexistent-id")
+
+
+class TestUpdateJobProgressUseCase:
+    def test_update_running_job(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        job.mark_running()
+        repo.update(job)
+
+        progress_uc = UpdateJobProgressUseCase(repo)
+        updated = progress_uc.execute(
+            UpdateJobProgressCommand(job_id=job.id, progress=75.0, current_stage="渲染中")
+        )
+        assert updated.progress == 75.0
+        assert updated.current_stage == "渲染中"
+
+    def test_update_non_running_raises(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+
+        progress_uc = UpdateJobProgressUseCase(repo)
+        with pytest.raises(ValueError, match="只有 running 状态"):
+            progress_uc.execute(
+                UpdateJobProgressCommand(job_id=job.id, progress=50.0)
+            )
+
+
+class TestCompleteJobUseCase:
+    def test_complete_running_job(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        job.mark_running()
+        repo.update(job)
+
+        complete_uc = CompleteJobUseCase(repo)
+        completed = complete_uc.execute(
+            CompleteJobCommand(job_id=job.id, result={"url": "https://example.com/v.mp4"})
+        )
+        assert completed.status == JobStatus.SUCCESS
+        assert completed.progress == 100.0
+        assert completed.result == {"url": "https://example.com/v.mp4"}
+
+    def test_complete_non_running_raises(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        job.mark_running()
+        job.mark_success()  # 已经是终态
+
+        complete_uc = CompleteJobUseCase(repo)
+        with pytest.raises(ValueError, match="只有 running/pending 状态"):
+            complete_uc.execute(CompleteJobCommand(job_id=job.id))
+
+
+class TestFailJobUseCase:
+    def test_fail_job(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        job.mark_running()
+        repo.update(job)
+
+        fail_uc = FailJobUseCase(repo)
+        failed = fail_uc.execute(FailJobCommand(job_id=job.id, error_message="磁盘空间不足"))
+        assert failed.status == JobStatus.FAILED
+        assert failed.error_message == "磁盘空间不足"
+
+
+class TestRetryJobUseCase:
+    def test_retry_failed_job(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, max_retries=3))
+        job.mark_running()
+        job.mark_failed("网络超时")
+        repo.update(job)
+
+        retry_uc = RetryJobUseCase(repo)
+        retried = retry_uc.execute(job.id)
+        assert retried.status == JobStatus.PENDING
+        assert retried.retry_count == 1
+        assert retried.error_message == ""
+
+    def test_retry_non_failed_raises(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+
+        retry_uc = RetryJobUseCase(repo)
+        with pytest.raises(ValueError, match="任务不可重试"):
+            retry_uc.execute(job.id)
+
+
+class TestCancelJobUseCase:
+    def test_cancel_pending_job(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+
+        cancel_uc = CancelJobUseCase(repo)
+        cancelled = cancel_uc.execute(job.id)
+        assert cancelled.status == JobStatus.CANCELLED
+
+    def test_cancel_completed_raises(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        job.mark_running()
+        job.mark_success()
+        repo.update(job)
+
+        cancel_uc = CancelJobUseCase(repo)
+        with pytest.raises(ValueError, match="已处于终态"):
+            cancel_uc.execute(job.id)
+
+
+class TestGetJobUseCase:
+    def test_get_existing_job(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+
+        get_uc = GetJobUseCase(repo)
+        fetched = get_uc.execute(job.id)
+        assert fetched is not None
+        assert fetched.id == job.id
+
+    def test_get_nonexistent_returns_none(self, repo):
+        get_uc = GetJobUseCase(repo)
+        assert get_uc.execute("nonexistent") is None
+
+
+class TestListJobsUseCase:
+    def test_list_by_project(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.RENDER_EDIT_PLAN))
+        create_uc.execute(CreateJobCommand(project_id="proj-2", job_type=JobType.VIDEO_COMPOSE))
+
+        list_uc = ListJobsUseCase(repo)
+        jobs = list_uc.execute(project_id="proj-1")
+        assert len(jobs) == 2
+
+    def test_list_by_project_with_type_filter(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.RENDER_EDIT_PLAN))
+
+        list_uc = ListJobsUseCase(repo)
+        jobs = list_uc.execute(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)
+        assert len(jobs) == 1
+        assert jobs[0].job_type == JobType.VIDEO_COMPOSE
+
+    def test_list_by_project_with_status_filter(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        j1 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        j2 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        j2.mark_running()
+        repo.update(j2)
+
+        list_uc = ListJobsUseCase(repo)
+        running_jobs = list_uc.execute(project_id="proj-1", status=JobStatus.RUNNING)
+        assert len(running_jobs) == 1
+
+    def test_list_by_user(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, created_by_user_id="user-1"))
+        create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, created_by_user_id="user-2"))
+
+        list_uc = ListJobsUseCase(repo)
+        jobs = list_uc.execute(user_id="user-1")
+        assert len(jobs) == 1
+
+    def test_list_without_project_or_user_raises(self, repo):
+        list_uc = ListJobsUseCase(repo)
+        with pytest.raises(ValueError, match="必须指定"):
+            list_uc.execute()
+
+
+class TestGetJobStatisticsUseCase:
+    def test_statistics(self, repo):
+        create_uc = CreateJobUseCase(repo)
+        j1 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        j2 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        j2.mark_running()
+        repo.update(j2)
+        j3 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        j3.mark_running()
+        j3.mark_success()
+        repo.update(j3)
+        j4 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE))
+        j4.mark_running()
+        j4.mark_failed("err")
+        repo.update(j4)
+
+        stats_uc = GetJobStatisticsUseCase(repo)
+        stats = stats_uc.execute("proj-1")
+        assert stats["total"] == 4
+        assert stats["pending"] == 1
+        assert stats["running"] == 1
+        assert stats["success"] == 1
+        assert stats["failed"] == 1
+
+
+# ── JobService 集成测试 ───────────────────────────────────────────────────────
+
+
+class TestJobService:
+    def test_create_compose_job(self, repo):
+        from app.services.job_service import JobService
+
+        svc = JobService(repo)
+        job = svc.create_compose_job(
+            project_id="proj-1",
+            plan_id="plan-1",
+            user_id="user-1",
+        )
+        assert job.job_type == JobType.VIDEO_COMPOSE
+        assert job.payload == {"plan_id": "plan-1"}
+        assert job.source_id == "plan-1"
+        assert job.created_by_user_id == "user-1"
+
+    def test_create_render_job(self, repo):
+        from app.services.job_service import JobService
+
+        svc = JobService(repo)
+        job = svc.create_render_job(
+            project_id="proj-1",
+            plan_id="plan-1",
+            user_id="user-1",
+        )
+        assert job.job_type == JobType.RENDER_EDIT_PLAN
+
+    def test_submit_compose_if_not_exists_creates_new(self, repo):
+        from app.services.job_service import JobService
+
+        svc = JobService(repo)
+        job, created = svc.submit_compose_if_not_exists("proj-1", "plan-1", "user-1", "celery-1")
+        assert created is True
+        assert job.status == JobStatus.RUNNING
+        assert job.celery_task_id == "celery-1"
+
+    def test_submit_compose_if_not_exists_returns_existing(self, repo):
+        from app.services.job_service import JobService
+
+        svc = JobService(repo)
+        job1, created1 = svc.submit_compose_if_not_exists("proj-1", "plan-1", "user-1", "celery-1")
+        job2, created2 = svc.submit_compose_if_not_exists("proj-1", "plan-1", "user-1", "celery-2")
+        assert created2 is False
+        assert job2.id == job1.id
+
+    def test_full_lifecycle(self, repo):
+        """完整生命周期测试：创建 → 提交 → 更新进度 → 完成。"""
+        from app.services.job_service import JobService
+
+        svc = JobService(repo)
+
+        # 创建
+        job = svc.create_compose_job("proj-1", "plan-1", "user-1")
+        assert job.status == JobStatus.PENDING
+
+        # 提交
+        job = svc.submit_job(job.id, "celery-xyz")
+        assert job.status == JobStatus.RUNNING
+        assert job.celery_task_id == "celery-xyz"
+
+        # 更新进度
+        job = svc.update_progress(job.id, 50.0, "合成中")
+        assert job.progress == 50.0
+
+        job = svc.update_progress(job.id, 80.0, "上传结果")
+        assert job.progress == 80.0
+
+        # 完成
+        job = svc.complete_job(job.id, result={"url": "https://example.com/v.mp4"})
+        assert job.status == JobStatus.SUCCESS
+        assert job.progress == 100.0
+
+    def test_full_lifecycle_with_retry(self, repo):
+        """带重试的完整生命周期测试。"""
+        from app.services.job_service import JobService
+
+        svc = JobService(repo)
+
+        # 创建并执行
+        job = svc.create_compose_job("proj-1", "plan-1", "user-1", max_retries=2)
+        job = svc.submit_job(job.id, "celery-1")
+
+        # 失败
+        job = svc.fail_job(job.id, "临时网络错误")
+        assert job.status == JobStatus.FAILED
+
+        # 重试
+        job = svc.retry_job(job.id)
+        assert job.status == JobStatus.PENDING
+        assert job.retry_count == 1
+
+        # 再次提交
+        job = svc.submit_job(job.id, "celery-2")
+        assert job.status == JobStatus.RUNNING
+
+        # 成功
+        job = svc.complete_job(job.id, result={"url": "ok"})
+        assert job.status == JobStatus.SUCCESS