alembic/versions/043_add_updated_at_to_generation_tasks.py

@@ -0,0 +1,33 @@
+"""add updated_at to generation_tasks
+
+Revision ID: 043_updated_at_generation_tasks
+Revises: 042_storage_key
+Create Date: 2026-07-18 19:30:00.000000
+
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "043_updated_at_generation_tasks"
+down_revision = "042_storage_key"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "generation_tasks",
+        sa.Column(
+            "updated_at",
+            sa.DateTime,
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("generation_tasks", "updated_at")

apps/api/app/api/routes/_helpers.py

@@ -1,5 +1,6 @@
 """路由层共享辅助函数 — 消除跨文件重复定义。"""
 
+from datetime import datetime, timezone
 from typing import Any
 
 from fastapi import HTTPException, status
@@ -122,3 +123,19 @@ def auto_select_video_assets(
     # 按创建时间降序（新素材在前）
     filtered_videos.sort(key=lambda a: a.created_at, reverse=True)
     return [a.id for a in filtered_videos]
+
+
+def format_utc_datetime(dt: datetime | None) -> str:
+    """将数据库读出的 UTC naive datetime 格式化为带时区的 ISO 8601 字符串。
+
+    数据库 DateTime 列不带时区信息，但存的是 UTC 时间。
+    直接 .isoformat() 输出无时区标识，前端会按本地时间解析，导致差 8 小时。
+    输出带 Z 后缀，前端 new Date() 自动转本地时间。
+    """
+    if dt is None:
+        return ""
+    if isinstance(dt, str):
+        return dt
+    if dt.tzinfo is None:
+        return dt.isoformat() + "Z"
+    return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")

apps/api/app/api/routes/assets.py

@@ -1,7 +1,7 @@
 import logging
 from typing import Any, Optional
 
-from app.api.routes._helpers import check_project_access
+from app.api.routes._helpers import check_project_access, format_utc_datetime
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.storage import get_storage_service
 from app.dependencies import (
@@ -71,7 +71,7 @@ def _to_asset_response(item, storage_service=None) -> AssetResponse:
         status=item.status.value,
         classification_status=item.classification_status.value,
         quality_score=item.quality_score,
-        created_at=item.created_at.isoformat() if item.created_at else None,
+        created_at=format_utc_datetime(item.created_at),
         uploaded_by_user_id=item.uploaded_by_user_id,
         tag_ids=getattr(item, "tag_ids", []),
     )

apps/api/app/api/routes/task_center.py

@@ -1,6 +1,7 @@
 import logging
 from typing import Any
 
+from app.api.routes._helpers import format_utc_datetime
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.celery_app import celery_app
 from app.core.task_enqueue import (
@@ -99,8 +100,8 @@ def _generation_task_to_user_response(task) -> UserTaskResponse:
         retryable=_status_value(task.status) == "failed",
         retry_count=task.retry_count or 0,
         source_id=task.id,
-        created_at=task.created_at,
-        updated_at=task.completed_at or task.started_at or task.created_at,
+        created_at=format_utc_datetime(task.created_at),
+        updated_at=format_utc_datetime(task.completed_at or task.started_at or task.created_at),
     )
 
 
@@ -119,8 +120,8 @@ def _generation_task_to_project_response(task) -> ProjectTaskResponse:
         retry_count=task.retry_count or 0,
         source_id=task.id,
         template_id=task.template_id,
-        created_at=task.created_at,
-        updated_at=task.completed_at or task.started_at or task.created_at,
+        created_at=format_utc_datetime(task.created_at),
+        updated_at=format_utc_datetime(task.completed_at or task.started_at or task.created_at),
     )
 
 
@@ -284,8 +285,8 @@ def list_project_tasks(
                     user_message=_humanize_task_error(job.error_message),
                     retryable=_status_value(job.status) == "failed",
                     source_id=job.id,
-                    created_at=job.created_at,
-                    updated_at=job.updated_at,
+                    created_at=format_utc_datetime(job.created_at),
+                    updated_at=format_utc_datetime(job.updated_at),
                 )
             )
 
@@ -383,7 +384,7 @@ def retry_project_task(
             progress=0,
             current_step=_ingest_step(retried),
             source_id=retried.id,
-            created_at=retried.created_at,
-            updated_at=retried.updated_at,  # type: ignore[attr-defined]
+            created_at=format_utc_datetime(retried.created_at),
+            updated_at=format_utc_datetime(retried.updated_at),  # type: ignore[attr-defined]
         )
     raise HTTPException(status_code=400, detail="Unsupported task type")

apps/api/app/api/routes/videos.py

@@ -1,5 +1,6 @@
 import logging
 
+from app.api.routes._helpers import format_utc_datetime
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.celery_app import celery_app
 from app.core.storage import OSSStorageService, get_storage_service
@@ -50,7 +51,7 @@ def _to_video_response(item, storage: OSSStorageService | None = None) -> VideoI
         review_status=item.review_status,
         generation_params=item.generation_params,
         download_url=download_url,
-        generated_at=item.generated_at.isoformat() if hasattr(item, "generated_at") and item.generated_at else "",
+        generated_at=format_utc_datetime(item.generated_at) if hasattr(item, "generated_at") else "",
     )
 
 

apps/api/app/schemas/task_center.py

@@ -1,5 +1,3 @@
-from datetime import datetime
-
 from pydantic import BaseModel, Field
 
 
@@ -17,8 +15,8 @@ class ProjectTaskResponse(BaseModel):
     retry_count: int = 0
     source_id: str = ""
     template_id: str = ""
-    created_at: datetime | None = None
-    updated_at: datetime | None = None
+    created_at: str = ""
+    updated_at: str = ""
 
 
 class ListProjectTasksResponse(BaseModel):
@@ -42,8 +40,8 @@ class UserTaskResponse(BaseModel):
     retryable: bool = False
     retry_count: int = 0
     source_id: str = ""
-    created_at: datetime | None = None
-    updated_at: datetime | None = None
+    created_at: str = ""
+    updated_at: str = ""
 
 
 class ListTasksResponse(BaseModel):

apps/worker/worker_app/celery_app.py

@@ -17,5 +17,6 @@ celery_app.conf.imports = (
     "worker_app.tasks.edit_plan_generation",
     "worker_app.tasks.compose_video",
     "worker_app.tasks.batch_download",
+    "worker_app.tasks._startup",
     "apps.worker.video_processing.dedup",
 )

apps/worker/worker_app/tasks/_startup.py

@@ -0,0 +1,55 @@
+"""Worker 启动时的初始化任务 — 孤儿任务清理等。"""
+
+import logging
+
+from worker_app.celery_app import celery_app
+from worker_app.db import SessionLocal
+
+logger = logging.getLogger(__name__)
+
+ORPHAN_TASK_TIMEOUT_MINUTES = 10
+
+
+def cleanup_orphan_tasks(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> int:
+    """清理数据库中超时未更新的 running 任务。
+
+    worker 重启或崩溃后，之前处于 running 状态的任务会变成孤儿任务，
+    一直卡在 running 不动。通过 updated_at 超时判断并标记为 failed。
+
+    Args:
+        timeout_minutes: 超时时间（分钟），默认 10 分钟
+
+    Returns:
+        清理的任务数量
+    """
+    from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+        SQLAlchemyGenerationTaskRepository,
+    )
+
+    try:
+        session = SessionLocal()
+        repo = SQLAlchemyGenerationTaskRepository(session)
+        count = repo.cleanup_stale_running(timeout_minutes)
+        session.close()
+        if count > 0:
+            logger.warning("清理了 %d 个超时的孤儿 running 任务", count)
+        else:
+            logger.info("无孤儿 running 任务需要清理")
+        return count
+    except Exception as e:
+        logger.error("清理孤儿任务失败: %s", e, exc_info=True)
+        return 0
+
+
+@celery_app.on_after_configure.connect
+def _setup_periodic_tasks(sender, **kwargs):
+    """Celery 配置完成后，注册 worker 启动钩子。"""
+    pass
+
+
+@celery_app.on_worker_ready.connect
+def _on_worker_ready(sender, **kwargs):
+    """Worker 启动完成后执行 — 清理孤儿任务。"""
+    logger.info("Worker 启动完成，开始清理孤儿 running 任务...")
+    count = cleanup_orphan_tasks()
+    logger.info("Worker 启动清理完成，共清理 %d 个孤儿任务", count)

apps/worker/worker_app/tasks/generation.py

@@ -1409,6 +1409,16 @@ def generate_video(self, task_id: str) -> dict:
 
     logger.info("[task_id=%s] [接收任务] 开始生成视频任务", task_id)
 
+    # 调度时清理一次孤儿任务（其他 worker 崩溃留下的 running 任务）
+    try:
+        from worker_app.tasks._startup import cleanup_orphan_tasks
+
+        orphan_count = cleanup_orphan_tasks()
+        if orphan_count > 0:
+            logger.info("[task_id=%s] 调度前清理了 %d 个孤儿任务", task_id, orphan_count)
+    except Exception:
+        pass
+
     # ── 1. 加载任务信息 ──────────────────────────────────────────────────────
     task_info = _load_task_info(task_id)
     if task_info is None:

docs/schema-metadata-snapshot.json

@@ -1628,6 +1628,14 @@
           "primary_key": false,
           "type": "DATETIME",
           "unique": false
+        },
+        {
+          "index": false,
+          "name": "updated_at",
+          "nullable": false,
+          "primary_key": false,
+          "type": "DATETIME",
+          "unique": false
         }
       ],
       "indexes": [

packages/adapters/sqlalchemy_impl/generation_task_repository.py

@@ -1,3 +1,5 @@
+from datetime import datetime, timezone
+
 from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
@@ -33,6 +35,7 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
         batch_id=model.batch_id or "",
         logs=model.logs or "[]",
         created_at=model.created_at,
+        updated_at=model.updated_at,
     )
 
 
@@ -67,6 +70,7 @@ class SQLAlchemyGenerationTaskRepository:
             batch_id=task.batch_id or "",
             logs=task.logs,
             created_at=task.created_at,
+            updated_at=task.updated_at,
         )
         self.session.add(model)
         self.session.commit()
@@ -225,3 +229,37 @@ class SQLAlchemyGenerationTaskRepository:
         model.logs = task.logs
         self.session.commit()
         return task
+
+    def cleanup_stale_running(self, timeout_minutes: int = 10) -> int:
+        """清理超时未更新的 running 任务（孤儿任务）。
+
+        将 status=running 且 updated_at 超过 timeout_minutes 分钟未更新的任务
+        标记为 failed，error_message 标记为任务执行中断。
+
+        Returns:
+            清理的任务数量
+        """
+        from datetime import timedelta
+
+        cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes)
+        models = (
+            self.session.query(GenerationTaskModel)
+            .filter(
+                GenerationTaskModel.status == GenerationTaskStatus.RUNNING.value,
+                GenerationTaskModel.updated_at < cutoff,
+            )
+            .all()
+        )
+        if not models:
+            return 0
+        for model in models:
+            model.status = GenerationTaskStatus.FAILED.value
+            model.error_message = "任务执行中断（worker重启/超时）"
+            model.error_info = {
+                "error_type": "WorkerInterrupted",
+                "message": "任务在运行中中断，可能因 worker 重启或超时",
+                "failed_at": datetime.now(timezone.utc).isoformat(),
+            }
+            model.completed_at = datetime.now(timezone.utc)
+        self.session.commit()
+        return len(models)

packages/adapters/sqlalchemy_impl/models.py

@@ -269,6 +269,12 @@ class GenerationTaskModel(Base):
     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))
+    updated_at = Column(
+        DateTime,
+        nullable=False,
+        default=lambda: datetime.now(timezone.utc),
+        onupdate=lambda: datetime.now(timezone.utc),
+    )
 
 
 class GeneratedVideoModel(Base):

packages/domain/generation_task.py

@@ -92,6 +92,7 @@ class GenerationTask:
     batch_id: str = ""
     logs: str = "[]"
     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(

tests/unit/test_generation_task_orphan_cleanup.py

@@ -0,0 +1,126 @@
+"""GenerationTaskRepository - cleanup_stale_running 孤儿任务清理单元测试。"""
+
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+    SQLAlchemyGenerationTaskRepository,
+)
+from packages.adapters.sqlalchemy_impl.models import Base
+from packages.domain import GenerationTask, GenerationTaskStatus
+
+
+def _repository():
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine)()
+    return SQLAlchemyGenerationTaskRepository(session), session, engine
+
+
+def _make_task(**kwargs) -> GenerationTask:
+    defaults = dict(
+        project_id="proj-1",
+        asset_library_id="lib-1",
+        created_by_user_id="user-1",
+    )
+    defaults.update(kwargs)
+    return GenerationTask.create(**defaults)
+
+
+def test_cleanup_stale_running_no_tasks_returns_zero():
+    """没有任务时返回0。"""
+    repo, _, _ = _repository()
+    count = repo.cleanup_stale_running(timeout_minutes=10)
+    assert count == 0
+
+
+def test_cleanup_stale_running_recent_running_not_cleaned():
+    """刚启动的 running 任务不清理。"""
+    repo, _, _ = _repository()
+    task = _make_task()
+    repo.create(task)
+    task.mark_processing()
+    repo.update(task)
+
+    count = repo.cleanup_stale_running(timeout_minutes=10)
+    assert count == 0
+    assert repo.get(task.id).status == GenerationTaskStatus.RUNNING
+
+
+def test_cleanup_stale_running_old_running_marked_failed():
+    """超时的 running 任务被标记为 failed。"""
+    repo, _, engine = _repository()
+    task = _make_task()
+    repo.create(task)
+    task.mark_processing()
+    repo.update(task)
+
+    # 手动把 updated_at 改到 30 分钟前
+    with engine.connect() as conn:
+        conn.execute(
+            text("UPDATE generation_tasks SET updated_at = :ts WHERE id = :id"),
+            {"ts": datetime.now(timezone.utc) - timedelta(minutes=30), "id": task.id},
+        )
+        conn.commit()
+
+    count = repo.cleanup_stale_running(timeout_minutes=10)
+    assert count == 1
+
+    saved = repo.get(task.id)
+    assert saved.status == GenerationTaskStatus.FAILED
+    assert "任务执行中断" in saved.error_message
+    assert saved.error_info.get("error_type") == "WorkerInterrupted"
+    assert saved.completed_at is not None
+
+
+def test_cleanup_stale_running_pending_not_touched():
+    """pending 状态即使超时也不清理。"""
+    repo, _, engine = _repository()
+    task = _make_task()
+    repo.create(task)
+
+    with engine.connect() as conn:
+        conn.execute(
+            text("UPDATE generation_tasks SET updated_at = :ts WHERE id = :id"),
+            {"ts": datetime.now(timezone.utc) - timedelta(hours=1), "id": task.id},
+        )
+        conn.commit()
+
+    count = repo.cleanup_stale_running(timeout_minutes=10)
+    assert count == 0
+    assert repo.get(task.id).status == GenerationTaskStatus.PENDING
+
+
+def test_cleanup_stale_running_multiple_orphans():
+    """多个超时 running 任务全部清理。"""
+    repo, _, engine = _repository()
+
+    tasks = []
+    for i in range(3):
+        t = _make_task(project_id=f"proj-{i}")
+        repo.create(t)
+        t.mark_processing()
+        repo.update(t)
+        tasks.append(t)
+
+    # 前两个超时，第三个是新的
+    with engine.connect() as conn:
+        for t in tasks[:2]:
+            conn.execute(
+                text("UPDATE generation_tasks SET updated_at = :ts WHERE id = :id"),
+                {"ts": datetime.now(timezone.utc) - timedelta(minutes=20), "id": t.id},
+            )
+        conn.commit()
+
+    count = repo.cleanup_stale_running(timeout_minutes=10)
+    assert count == 2
+
+    assert repo.get(tasks[0].id).status == GenerationTaskStatus.FAILED
+    assert repo.get(tasks[1].id).status == GenerationTaskStatus.FAILED
+    assert repo.get(tasks[2].id).status == GenerationTaskStatus.RUNNING