fix: 数据处理状态&数据评估

runtime/datamate-python/app/core/exception/codes.py

@@ -92,6 +92,12 @@ def __init__(self):
     EVALUATION_MODEL_NOT_FOUND: Final = ErrorCode(
         "evaluation.0003", "Evaluation model not found", 404
     )
+    EVALUATION_NO_ITEMS: Final = ErrorCode(
+        "evaluation.0004", "No evaluable items found in the selected data source", 400
+    )
+    EVALUATION_ALL_ITEMS_FAILED: Final = ErrorCode(
+        "evaluation.0005", "All evaluation items failed to be evaluated", 500
+    )
 
     # ========== 生成模块 ==========
     GENERATION_TASK_NOT_FOUND: Final = ErrorCode(

runtime/datamate-python/app/module/cleaning/repository/cleaning_task_repository.py

@@ -1,10 +1,20 @@
+from datetime import datetime
 from typing import List, Optional
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy import select, delete, func
+from sqlalchemy import select, delete, func, update
 from app.db.models.cleaning import CleaningTask
 from app.module.cleaning.schema import CleaningTaskDto
 
 
+# 终态集合：写入终态后不再接受控制面的状态覆盖
+TERMINAL_TASK_STATUSES = (
+    "COMPLETED",
+    "PARTIAL_SUCCESS",
+    "FAILED",
+    "STOPPED",
+)
+
+
 class CleaningTaskRepository:
     """Repository for cleaning task operations"""
 
@@ -127,6 +137,42 @@ async def update_task(self, db: AsyncSession, task: CleaningTaskDto) -> None:
 
             await db.flush()
 
+    async def find_status(self, db: AsyncSession, task_id: str) -> Optional[str]:
+        """Query task status only (cheap read for polling)"""
+        query = select(self.model.status).where(self.model.id == task_id)
+        result = await db.execute(query)
+        return result.scalar_one_or_none()
+
+    async def finish_if_unfinished(
+        self,
+        db: AsyncSession,
+        task_id: str,
+        status: str,
+        finished_at: Optional[datetime] = None
+    ) -> bool:
+        """Write a terminal status only when the task is not terminal yet.
+
+        Returns True when the row was updated, False when the task was missing
+        or already in a terminal state. This keeps the control plane from
+        overwriting a status the executor already wrote directly to the table.
+        """
+        values = {"status": status}
+        if finished_at is not None:
+            values["finished_at"] = finished_at
+
+        query = (
+            update(self.model)
+            .where(
+                self.model.id == task_id,
+                self.model.status.notin_(TERMINAL_TASK_STATUSES),
+            )
+            .values(**values)
+            .execution_options(synchronize_session=False)
+        )
+        result = await db.execute(query)
+        await db.flush()
+        return bool(result.rowcount and result.rowcount > 0)
+
     async def delete_task_by_id(self, db: AsyncSession, task_id: str) -> None:
         """Delete task by ID"""
         query = delete(self.model).where(self.model.id == task_id)

runtime/datamate-python/app/module/cleaning/runtime_client.py

@@ -4,6 +4,10 @@
 
 logger = get_logger(__name__)
 
+# 状态查询是轻量的本地接口，不需要沿用提交任务的 60s 超时；
+# 超时只会让这一轮轮询“无结论”，不会再据此判失败。
+STATUS_REQUEST_TIMEOUT_SECONDS = 10.0
+
 
 class RuntimeClient:
     """HTTP client for communicating with runtime service"""
@@ -46,7 +50,7 @@ async def get_task_status(self, task_id: str) -> Optional[dict]:
         """Get task status from runtime"""
         try:
             url = f"{self.base_url}/api/task/{task_id}/status"
-            response = await self.client.get(url)
+            response = await self.client.get(url, timeout=STATUS_REQUEST_TIMEOUT_SECONDS)
             response.raise_for_status()
             return response.json()
         except httpx.HTTPError as e: