Merge branch 'codex/fix-running-task-timeout-from-current'

app/services/task_service.py

@@ -21,6 +21,7 @@ from app.services.report_service import generate_hotspot_report, generate_item_r
 RUNNING_TASK_MESSAGE = "当前有正在运行的任务，请稍后再试"
 RESTART_ERROR_MESSAGE = "系统重启，任务被中断"
 STALE_PROGRESS_THRESHOLD_SECONDS = 10 * 60
+STALE_PROGRESS_ERROR_TYPE = "stale_progress_timeout"
 task_executor = ThreadPoolExecutor(max_workers=1)
 
 
@@ -44,6 +45,30 @@ def has_running_task(session: Session) -> bool:
     return session.scalar(select(Task.id).where(Task.status == "running").limit(1)) is not None
 
 
+def recover_stale_running_tasks(session: Session) -> int:
+    now = ensure_utc_datetime(utc_now())
+    tasks = list(session.scalars(select(Task).where(Task.status == "running")))
+    recovered = 0
+    for task in tasks:
+        last_progress_at = ensure_utc_datetime(task.last_progress_at or task.started_at or task.created_at)
+        if last_progress_at is None or now is None:
+            continue
+        seconds_since_progress = max(0, int((now - last_progress_at).total_seconds()))
+        if seconds_since_progress < STALE_PROGRESS_THRESHOLD_SECONDS:
+            continue
+        _mark_task_failed(
+            task,
+            "system",
+            STALE_PROGRESS_ERROR_TYPE,
+            f"任务超过 {STALE_PROGRESS_THRESHOLD_SECONDS // 60} 分钟没有进度更新，已自动标记失败",
+        )
+        task.finished_at = task.finished_at or utc_now()
+        recovered += 1
+    if recovered:
+        session.commit()
+    return recovered
+
+
 def recover_running_tasks(session: Session) -> int:
     tasks = list(session.scalars(select(Task).where(Task.status == "running")))
     for task in tasks:
@@ -256,6 +281,8 @@ def run_task(task_id: str, *, session_factory: Callable[[], Session] | sessionma
         task = session.get(Task, task_id)
         if task is None:
             return
+        task.started_at = task.started_at or utc_now()
+        session.commit()
         try:
             platform = build_platform(task.platform)
             ai_requester, report_summary_provider = build_ai_dependencies()
@@ -345,6 +372,7 @@ def run_task(task_id: str, *, session_factory: Callable[[], Session] | sessionma
                             [{"comment_id": comment.id, "content": comment.content} for comment in item_comments],
                             requester=ai_requester,
                             max_retries=get_settings().ai_max_retries,
+                            timeout_seconds=get_settings().ai_timeout_seconds,
                         )
                         results_by_comment_id = {result.comment_id: result for result in ai_results}
                         for comment in item_comments:
@@ -390,9 +418,11 @@ def run_task(task_id: str, *, session_factory: Callable[[], Session] | sessionma
                 update_task_stage(task, "success")
             else:
                 _mark_task_failed(task, task.error_stage or "crawl_items", task.error_type or "no_successful_items", task.error_message or "没有任何内容条目成功")
+            task.finished_at = task.finished_at or utc_now()
             session.commit()
         except Exception as exc:
             _mark_task_failed(task, "system", "unexpected_error", str(exc))
+            task.finished_at = task.finished_at or utc_now()
             session.commit()
     finally:
         if close_session: