feat(data-process): 完善后台生成与失败重试

backend/app/modules/data_process/generation.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import hashlib
 import json
+import logging
 import re
 from collections.abc import Callable, Iterable, Mapping
 from typing import Any
@@ -22,6 +23,10 @@ class ModelGenerationError(ValueError):
     """模型配置、响应或调用失败。"""
 
 
+class _TerminalModelGenerationError(ModelGenerationError):
+    """使用相同参数重试也无法恢复的模型响应错误。"""
+
+
 OUTPUT_TYPE_STANDARD = "standard"
 OUTPUT_TYPE_REASONING = "reasoning"
 SUPPORTED_OUTPUT_TYPES = {OUTPUT_TYPE_STANDARD, OUTPUT_TYPE_REASONING}
@@ -31,6 +36,25 @@ SUPPORTED_REASONING_DETAILS = {
     REASONING_DETAIL_NORMAL,
     REASONING_DETAIL_DETAILED,
 }
+MINIMAX_M3_API_HOSTS = {"api.minimax.io", "api.minimaxi.com"}
+MINIMAX_M3_MIN_COMPLETION_TOKENS = 4096
+logger = logging.getLogger(__name__)
+
+
+def _is_retryable_generation_error(exc: Exception) -> bool:
+    if isinstance(exc, _TerminalModelGenerationError):
+        return False
+    if isinstance(exc, httpx.HTTPStatusError):
+        status_code = exc.response.status_code
+        return status_code in {408, 425, 429} or status_code >= 500
+    if isinstance(exc, httpx.RequestError):
+        return True
+    return isinstance(exc, (json.JSONDecodeError, ModelGenerationError))
+
+
+def _is_official_minimax_m3(endpoint: str, model_name: str) -> bool:
+    host = (urlsplit(endpoint).hostname or "").casefold()
+    return host in MINIMAX_M3_API_HOSTS and model_name.casefold() == "minimax-m3"
 
 
 def chat_completions_url(value: str) -> str:
@@ -59,32 +83,139 @@ def chat_completions_url(value: str) -> str:
     return urlunsplit((parsed.scheme, parsed.netloc, target_path, "", ""))
 
 
-def _message_content(payload: Mapping[str, Any]) -> str:
+def _response_choice(payload: Mapping[str, Any]) -> Mapping[str, Any]:
     try:
-        content = payload["choices"][0]["message"]["content"]
+        choice = payload["choices"][0]
     except (KeyError, IndexError, TypeError) as exc:
-        raise ModelGenerationError(
-            "model response does not contain choices[0].message.content"
-        ) from exc
+        raise ModelGenerationError("模型响应缺少 choices[0]") from exc
+    if not isinstance(choice, Mapping):
+        raise ModelGenerationError("模型响应 choices[0] 不是对象")
+    return choice
+
+
+def _response_finish_reason(payload: Mapping[str, Any]) -> str:
+    try:
+        return str(_response_choice(payload).get("finish_reason") or "").strip().lower()
+    except ModelGenerationError:
+        return ""
+
+
+def _response_content_length(payload: Mapping[str, Any]) -> int:
+    try:
+        message = _response_choice(payload).get("message")
+        if not isinstance(message, Mapping):
+            return 0
+        content = message.get("content")
+        if isinstance(content, str):
+            return len(content)
+        if isinstance(content, list):
+            return sum(
+                len(str(item.get("text") or ""))
+                for item in content
+                if isinstance(item, Mapping)
+            )
+    except ModelGenerationError:
+        pass
+    return 0
+
+
+def _raise_for_terminal_response(payload: Mapping[str, Any]) -> Mapping[str, Any]:
+    choice = _response_choice(payload)
+    base_response = payload.get("base_resp")
+    status_code: Any = None
+    status_message = ""
+    if isinstance(base_response, Mapping):
+        status_code = base_response.get("status_code")
+        status_message = re.sub(
+            r"\s+", " ", str(base_response.get("status_msg") or "")
+        ).strip()[:200]
+
+    if bool(payload.get("input_sensitive")) or status_code in {1026, "1026"}:
+        raise _TerminalModelGenerationError(
+            f"模型输入触发内容安全拦截（code={status_code or 1026}）"
+        )
+    if bool(payload.get("output_sensitive")) or status_code in {1027, "1027"}:
+        raise _TerminalModelGenerationError(
+            f"模型输出触发内容安全拦截（code={status_code or 1027}）"
+        )
+
+    finish_reason = str(choice.get("finish_reason") or "").strip().lower()
+    if finish_reason == "length":
+        raise _TerminalModelGenerationError(
+            "模型输出因达到 Token 上限被截断（finish_reason=length），"
+            "请提高最大输出长度后重试"
+        )
+    if finish_reason == "content_filter":
+        raise _TerminalModelGenerationError(
+            "模型输出被内容安全策略拦截（finish_reason=content_filter）"
+        )
+    if finish_reason in {"tool_calls", "function_call"}:
+        raise _TerminalModelGenerationError(
+            f"模型返回了当前生成任务不支持的工具调用（finish_reason={finish_reason}）"
+        )
+    if status_code not in {None, "", 0, "0"}:
+        detail = f"：{status_message}" if status_message else ""
+        raise _TerminalModelGenerationError(
+            f"模型服务返回业务错误（code={status_code}）{detail}"
+        )
+    return choice
+
+
+def _message_content(payload: Mapping[str, Any]) -> str:
+    choice = _raise_for_terminal_response(payload)
+    message = choice.get("message")
+    if not isinstance(message, Mapping):
+        raise ModelGenerationError("模型响应缺少 choices[0].message")
+    content = message.get("content")
     if isinstance(content, str):
-        return content
-    if isinstance(content, list):
+        result = content
+    elif isinstance(content, list):
         parts = [
             str(item.get("text") or "")
             for item in content
             if isinstance(item, Mapping) and item.get("type") in {None, "text", "output_text"}
         ]
-        if parts:
-            return "".join(parts)
-    raise ModelGenerationError("model response content must be text")
+        result = "".join(parts)
+    elif content is None:
+        result = ""
+    else:
+        raise ModelGenerationError("模型响应 content 必须是文本")
+    if not result.strip():
+        raise ModelGenerationError("模型返回的最终内容为空，未生成可解析的 JSON")
+    return result
+
+
+def _json_documents(content: str) -> list[Any]:
+    decoder = json.JSONDecoder()
+    documents: list[Any] = []
+    cursor = 0
+    while cursor < len(content):
+        match = re.search(r"[\[{]", content[cursor:])
+        if not match:
+            break
+        start = cursor + match.start()
+        try:
+            value, end = decoder.raw_decode(content[start:])
+        except json.JSONDecodeError:
+            cursor = start + 1
+            continue
+        if isinstance(value, (Mapping, list)):
+            documents.append(value)
+        cursor = start + max(end, 1)
+    return documents
 
 
 def _json_payload(content: str) -> Any:
     # 只移除模型在 JSON 之前自行输出的思考过程，不能破坏 JSON 字段中的训练内容。
+    cleaned = content.strip()
+    if re.match(r"^\s*<think>", cleaned, flags=re.IGNORECASE) and not re.match(
+        r"^\s*<think>[\s\S]*?</think>", cleaned, flags=re.IGNORECASE
+    ):
+        raise ModelGenerationError("模型思考内容未闭合，响应可能已被截断")
     cleaned = re.sub(
-        r"^\s*<think>[\s\S]*?</think>\s*",
+        r"^\s*(?:<think>[\s\S]*?</think>\s*)+",
         "",
-        content,
+        cleaned,
         count=1,
         flags=re.IGNORECASE,
     ).strip()
@@ -93,10 +224,16 @@ def _json_payload(content: str) -> Any:
         cleaned = fenced.group(1).strip()
     try:
         return json.loads(cleaned)
-    except json.JSONDecodeError as exc:
+    except json.JSONDecodeError as direct_error:
+        documents = _json_documents(cleaned)
+        if len(documents) == 1:
+            return documents[0]
+        if len(documents) > 1:
+            raise ModelGenerationError("模型响应包含多个 JSON 对象，无法确定应使用哪一个")
         raise ModelGenerationError(
-            f"model response is not valid JSON at line {exc.lineno}, column {exc.colno}"
-        ) from exc
+            "模型响应中没有找到唯一且完整的 JSON 对象"
+            f"（第 {direct_error.lineno} 行，第 {direct_error.colno} 列）"
+        ) from direct_error
 
 
 def _result_items(payload: Any) -> list[Mapping[str, Any]]:
@@ -206,6 +343,7 @@ def generate_model_records(
     model_name = str(model.get("online_model_name") or model.get("name") or "").strip()
     if not model_name:
         raise ModelGenerationError("generation model name is required")
+    is_minimax_m3 = _is_official_minimax_m3(endpoint, model_name)
 
     temperature = float(config.get("temperature", 0.7))
     max_tokens = int(config.get("max_tokens", 1024))
@@ -246,9 +384,18 @@ def generate_model_records(
                         reasoning_detail=reasoning_detail,
                     ),
                     "temperature": temperature,
-                    "max_tokens": max_tokens,
                 }
-                if bool(config.get("json_mode", False)):
+                if is_minimax_m3:
+                    request_payload.update(
+                        reasoning_split=True,
+                        max_completion_tokens=max(
+                            max_tokens,
+                            MINIMAX_M3_MIN_COMPLETION_TOKENS,
+                        ),
+                    )
+                else:
+                    request_payload["max_tokens"] = max_tokens
+                if bool(config.get("json_mode", False)) and not is_minimax_m3:
                     request_payload["response_format"] = {"type": "json_object"}
 
                 last_error: Exception | None = None
@@ -264,7 +411,24 @@ def generate_model_records(
                         body = response.json()
                         if not isinstance(body, Mapping):
                             raise ModelGenerationError("model response body must be a JSON object")
-                        candidate_items = _result_items(_json_payload(_message_content(body)))
+                        try:
+                            candidate_items = _result_items(
+                                _json_payload(_message_content(body))
+                            )
+                        except ModelGenerationError as exc:
+                            logger.warning(
+                                "data process model response rejected task_id=%s model=%s "
+                                "finish_reason=%s response_chars=%s input_sensitive=%s "
+                                "output_sensitive=%s reason=%s",
+                                task_id,
+                                model_name,
+                                _response_finish_reason(body) or "missing",
+                                _response_content_length(body),
+                                bool(body.get("input_sensitive")),
+                                bool(body.get("output_sensitive")),
+                                str(exc),
+                            )
+                            raise
                         if len(candidate_items) < batch_count:
                             raise ModelGenerationError(
                                 "model response contains fewer result objects than requested: "
@@ -278,6 +442,8 @@ def generate_model_records(
                         ModelGenerationError,
                     ) as exc:
                         last_error = exc
+                        if not _is_retryable_generation_error(exc):
+                            break
 
                 if generated_items is None:
                     error_message = str(last_error or "model generation failed")[:2000]

backend/app/modules/data_process/schema_cli.py

@@ -7,6 +7,18 @@ from urllib.parse import urlsplit
 
 from app.modules.data_process.store import DataProcessStore
 
+REQUIRED_TASK_COLUMNS = (
+    "generation_run_id",
+    "results_confirmed",
+    "workflow_step",
+    "preview_status",
+    "preview_progress",
+    "preview_run_id",
+    "preview_failure_reason",
+    "preview_total_files",
+    "preview_completed_files",
+)
+
 
 def _target_label(database_url: str) -> str:
     parsed = urlsplit(database_url)
@@ -18,14 +30,13 @@ def _schema_ready(store: DataProcessStore) -> bool:
     with store.connect() as conn:
         row = conn.execute(
             """
-            SELECT EXISTS (
-                SELECT 1
-                FROM information_schema.columns
-                WHERE table_schema=current_schema()
-                  AND table_name='data_process_tasks'
-                  AND column_name='generation_run_id'
-            ) AS ready
-            """
+            SELECT COUNT(*) = %s AS ready
+            FROM information_schema.columns
+            WHERE table_schema=current_schema()
+              AND table_name='data_process_tasks'
+              AND column_name = ANY(%s)
+            """,
+            (len(REQUIRED_TASK_COLUMNS), list(REQUIRED_TASK_COLUMNS)),
         ).fetchone()
     return bool(row and row["ready"])
 

backend/app/modules/data_process/store.py

@@ -20,6 +20,8 @@ from app.modules.data_process.algorithms import estimate_token_count, stable_spl
 
 TASK_STATUSES = {"pending", "running", "completed", "failed", "stopped"}
 EDITABLE_STATUSES = {"pending", "failed", "stopped", "completed"}
+ACTIVE_PREVIEW_STATUSES = {"queued", "running"}
+WORKFLOW_STEPS = {"create", "model", "upload", "preview", "generate", "results"}
 
 _PREVIEW_CONFIG_ALIASES = {
     "preprocess_options": "preprocessOptions",
@@ -339,9 +341,10 @@ class DataProcessStore:
                     """
                     INSERT INTO data_process_tasks
                     (id, name, description, status, process_type, source_dataset_id, config,
-                     progress, tenant_id, project_id, owner_id, created_by, updated_by,
+                     progress, results_confirmed, tenant_id, project_id, owner_id, created_by, updated_by,
                      created_at, updated_at)
-                    VALUES (%s, %s, %s, 'pending', %s, %s, %s, 0, %s, %s, %s, %s, %s, %s, %s)
+                    VALUES (%s, %s, %s, 'pending', %s, %s, %s, 0, FALSE,
+                            %s, %s, %s, %s, %s, %s, %s)
                     RETURNING *
                     """,
                     (
@@ -442,9 +445,30 @@ class DataProcessStore:
     def _ensure_editable(task: dict[str, Any]) -> None:
         if task["status"] not in EDITABLE_STATUSES:
             raise InvalidStateError(f"task cannot be edited while status is {task['status']}")
+        if task.get("preview_status") in ACTIVE_PREVIEW_STATUSES:
+            raise InvalidStateError("task cannot be edited while preview is running")
         if task.get("output_dataset_id") and not _is_regeneration_prepared(task):
             raise InvalidStateError("published task cannot be edited")
 
+    def update_workflow_step(self, task_id: str, workflow_step: str) -> dict[str, Any]:
+        """独立保存向导位置，不触发配置或结果失效逻辑。"""
+
+        if workflow_step not in WORKFLOW_STEPS:
+            raise ValueError("invalid data process workflow step")
+        with self.connect() as conn:
+            row = conn.execute(
+                """
+                UPDATE data_process_tasks
+                SET workflow_step=%s, updated_at=%s
+                WHERE id=%s AND deleted_at IS NULL
+                RETURNING *
+                """,
+                (workflow_step, utcnow(), task_id),
+            ).fetchone()
+        if not row:
+            raise NotFoundError("data process task not found")
+        return _public_task(_decode_row(row)) or {}
+
     def recover_legacy_aborted_regeneration(self, task_id: str) -> dict[str, Any]:
         """恢复旧版在真正开始生成前误删的上一轮结果。
 
@@ -593,7 +617,8 @@ class DataProcessStore:
                     output_count=%s, filtered_count=0, duplicate_count=0,
                     error_count=(SELECT COUNT(*) FROM data_process_results
                                  WHERE task_id=%s AND status='invalid'),
-                    failure_reason=NULL, updated_at=%s
+                    failure_reason=NULL, results_confirmed=TRUE,
+                    workflow_step='results', updated_at=%s
                 WHERE id=%s
                 """,
                 (
@@ -660,6 +685,13 @@ class DataProcessStore:
                             "error_count": 0,
                             "failure_reason": None,
                             "generation_run_id": None,
+                            "results_confirmed": False,
+                            "preview_status": "idle",
+                            "preview_progress": 0,
+                            "preview_run_id": None,
+                            "preview_failure_reason": None,
+                            "preview_total_files": 0,
+                            "preview_completed_files": 0,
                             "started_at": None,
                             "completed_at": None,
                         }
@@ -709,6 +741,8 @@ class DataProcessStore:
                 task = self._task_in_connection(conn, task_id, for_update=True)
                 if task["status"] == "running":
                     raise ConflictError("running task cannot be prepared for regeneration")
+                if task.get("preview_status") in ACTIVE_PREVIEW_STATUSES:
+                    raise ConflictError("running preview cannot be prepared for regeneration")
 
                 current_updated_at = _serialize_value(task.get("updated_at"))
                 if payload["expected_updated_at"] != current_updated_at:
@@ -785,14 +819,19 @@ class DataProcessStore:
 
     def delete_task(self, task_id: str, *, deleted_by: str | None = None) -> None:
         with self.connect() as conn:
-            task = self._task_in_connection(conn, task_id, for_update=True)
-            if task["status"] == "running":
-                raise InvalidStateError("running task must be stopped before deletion")
+            self._task_in_connection(conn, task_id, for_update=True)
             now = utcnow()
             conn.execute(
                 """
                 UPDATE data_process_tasks
-                SET deleted_at=%s, deleted_by=%s, updated_at=%s
+                SET status=CASE WHEN status='running' THEN 'stopped' ELSE status END,
+                    generation_run_id=NULL,
+                    preview_status=CASE
+                        WHEN preview_status IN ('queued', 'running') THEN 'cancelled'
+                        ELSE preview_status
+                    END,
+                    preview_run_id=NULL,
+                    deleted_at=%s, deleted_by=%s, updated_at=%s
                 WHERE id=%s
                 """,
                 (now, deleted_by, now, task_id),
@@ -918,7 +957,10 @@ class DataProcessStore:
                             SELECT COALESCE(SUM(record_count), 0)
                             FROM data_process_source_files
                             WHERE task_id=%s AND deleted_at IS NULL
-                        ), updated_at=%s
+                        ), workflow_step='upload', preview_status='idle',
+                           preview_progress=0, preview_run_id=NULL,
+                           preview_failure_reason=NULL, preview_total_files=0,
+                           preview_completed_files=0, updated_at=%s
                         WHERE id=%s
                         """,
                         (task_id, now, task_id),
@@ -934,7 +976,12 @@ class DataProcessStore:
                         UPDATE data_process_tasks
                         SET status='pending', progress=%s, output_count=0, filtered_count=0,
                             duplicate_count=0, error_count=0, failure_reason=NULL,
-                            generation_run_id=NULL, started_at=NULL, completed_at=NULL,
+                            generation_run_id=NULL, results_confirmed=FALSE,
+                            workflow_step='upload', preview_status='idle',
+                            preview_progress=0, preview_run_id=NULL,
+                            preview_failure_reason=NULL, preview_total_files=0,
+                            preview_completed_files=0,
+                            started_at=NULL, completed_at=NULL,
                             input_count=(
                                 SELECT COALESCE(SUM(record_count), 0)
                                 FROM data_process_source_files
@@ -1035,7 +1082,10 @@ class DataProcessStore:
                     SET input_count=(SELECT COALESCE(SUM(record_count), 0)
                                      FROM data_process_source_files
                                      WHERE task_id=%s AND deleted_at IS NULL),
-                        updated_at=%s
+                        workflow_step='upload', preview_status='idle',
+                        preview_progress=0, preview_run_id=NULL,
+                        preview_failure_reason=NULL, preview_total_files=0,
+                        preview_completed_files=0, updated_at=%s
                     WHERE id=%s
                     """,
                     (task_id, now, task_id),
@@ -1049,6 +1099,11 @@ class DataProcessStore:
                     UPDATE data_process_tasks
                     SET status='pending', progress=0, output_count=0, filtered_count=0,
                         duplicate_count=0, error_count=0, failure_reason=NULL,
+                        results_confirmed=FALSE,
+                        workflow_step='upload', preview_status='idle',
+                        preview_progress=0, preview_run_id=NULL,
+                        preview_failure_reason=NULL, preview_total_files=0,
+                        preview_completed_files=0,
                         input_count=(SELECT COALESCE(SUM(record_count), 0)
                                      FROM data_process_source_files
                                      WHERE task_id=%s AND deleted_at IS NULL),
@@ -1064,6 +1119,7 @@ class DataProcessStore:
         items: Sequence[dict[str, Any]],
         *,
         source_file_ids: Sequence[str] | None = None,
+        preview_run_id: str | None = None,
     ) -> list[dict[str, Any]]:
         selected_ids = (
             list(dict.fromkeys(str(file_id) for file_id in source_file_ids))
@@ -1081,11 +1137,21 @@ class DataProcessStore:
             }
             if unexpected:
                 raise ValueError("preview items contain an unselected source file")
+        preview_file_count = len(selected_ids) if selected_ids is not None else len(
+            {str(item.get("source_file_id") or "") for item in items}
+        )
+        is_direct_build = preview_run_id is None
 
         now = utcnow()
         with self.connect() as conn:
             task = self._task_in_connection(conn, task_id, for_update=True)
-            self._ensure_editable(task)
+            if is_direct_build:
+                self._ensure_editable(task)
+            elif (
+                task.get("preview_run_id") != preview_run_id
+                or task.get("preview_status") != "running"
+            ):
+                raise InvalidStateError("preview run is no longer active")
             regeneration_prepared = _is_regeneration_prepared(task)
             if not regeneration_prepared:
                 conn.execute("DELETE FROM data_process_results WHERE task_id=%s", (task_id,))
@@ -1144,22 +1210,249 @@ class DataProcessStore:
                 ).fetchone()
                 created.append(_decode_row(row) or {})
             if regeneration_prepared:
-                conn.execute(
-                    "UPDATE data_process_tasks SET updated_at=%s WHERE id=%s",
-                    (now, task_id),
-                )
+                if is_direct_build:
+                    conn.execute(
+                        """
+                        UPDATE data_process_tasks
+                        SET workflow_step='preview', preview_status='completed',
+                            preview_progress=100, preview_run_id=NULL,
+                            preview_failure_reason=NULL, preview_total_files=%s,
+                            preview_completed_files=%s, updated_at=%s
+                        WHERE id=%s
+                        """,
+                        (preview_file_count, preview_file_count, now, task_id),
+                    )
+                else:
+                    conn.execute(
+                        "UPDATE data_process_tasks SET updated_at=%s WHERE id=%s",
+                        (now, task_id),
+                    )
             else:
                 conn.execute(
                     """
                     UPDATE data_process_tasks
                     SET status='pending', progress=20, output_count=0, filtered_count=0,
-                        duplicate_count=0, error_count=0, failure_reason=NULL, updated_at=%s
+                        duplicate_count=0, error_count=0, failure_reason=NULL,
+                        results_confirmed=FALSE,
+                        workflow_step=CASE WHEN %s THEN 'preview' ELSE workflow_step END,
+                        preview_status=CASE WHEN %s THEN 'completed' ELSE preview_status END,
+                        preview_progress=CASE WHEN %s THEN 100 ELSE preview_progress END,
+                        preview_run_id=CASE WHEN %s THEN NULL ELSE preview_run_id END,
+                        preview_failure_reason=CASE WHEN %s THEN NULL ELSE preview_failure_reason END,
+                        preview_total_files=CASE WHEN %s THEN %s ELSE preview_total_files END,
+                        preview_completed_files=CASE WHEN %s THEN %s ELSE preview_completed_files END,
+                        updated_at=%s
                     WHERE id=%s
                     """,
-                    (now, task_id),
+                    (
+                        is_direct_build,
+                        is_direct_build,
+                        is_direct_build,
+                        is_direct_build,
+                        is_direct_build,
+                        is_direct_build,
+                        preview_file_count,
+                        is_direct_build,
+                        preview_file_count,
+                        now,
+                        task_id,
+                    ),
                 )
         return created
 
+    def start_preview(
+        self,
+        task_id: str,
+        *,
+        source_file_ids: Sequence[str] | None = None,
+    ) -> tuple[dict[str, Any], list[str]]:
+        """创建一轮持久化切分任务，并返回本轮固定的源文件集合。"""
+
+        requested_ids = (
+            list(dict.fromkeys(str(file_id) for file_id in source_file_ids))
+            if source_file_ids is not None
+            else None
+        )
+        if requested_ids is not None and (
+            not requested_ids or any(not file_id for file_id in requested_ids)
+        ):
+            raise ValueError("source_file_ids must contain non-empty ids")
+
+        with self.connect() as conn:
+            task = self._task_in_connection(conn, task_id, for_update=True)
+            self._ensure_editable(task)
+            if requested_ids is None:
+                rows = conn.execute(
+                    """
+                    SELECT id FROM data_process_source_files
+                    WHERE task_id=%s AND deleted_at IS NULL
+                    ORDER BY created_at, id
+                    """,
+                    (task_id,),
+                ).fetchall()
+            else:
+                rows = conn.execute(
+                    """
+                    SELECT id FROM data_process_source_files
+                    WHERE task_id=%s AND deleted_at IS NULL AND id=ANY(%s)
+                    ORDER BY created_at, id
+                    """,
+                    (task_id, requested_ids),
+                ).fetchall()
+            selected_ids = [str(row["id"]) for row in rows]
+            if not selected_ids:
+                raise InvalidStateError("at least one source file is required")
+            if requested_ids is not None:
+                missing = set(requested_ids) - set(selected_ids)
+                if missing:
+                    raise NotFoundError(
+                        f"source files not found: {', '.join(sorted(missing))}"
+                    )
+
+            regeneration_prepared = _is_regeneration_prepared(task)
+            if not regeneration_prepared:
+                conn.execute("DELETE FROM data_process_results WHERE task_id=%s", (task_id,))
+            preview_run_id = new_id("dpprun")
+            now = utcnow()
+            row = conn.execute(
+                """
+                UPDATE data_process_tasks
+                SET status=CASE WHEN %s THEN status ELSE 'pending' END,
+                    progress=CASE WHEN %s THEN progress ELSE 0 END,
+                    output_count=CASE WHEN %s THEN output_count ELSE 0 END,
+                    filtered_count=CASE WHEN %s THEN filtered_count ELSE 0 END,
+                    duplicate_count=CASE WHEN %s THEN duplicate_count ELSE 0 END,
+                    error_count=CASE WHEN %s THEN error_count ELSE 0 END,
+                    failure_reason=CASE WHEN %s THEN failure_reason ELSE NULL END,
+                    results_confirmed=CASE WHEN %s THEN results_confirmed ELSE FALSE END,
+                    workflow_step='upload', preview_status='queued', preview_progress=0,
+                    preview_run_id=%s, preview_failure_reason=NULL,
+                    preview_total_files=%s, preview_completed_files=0,
+                    updated_at=%s
+                WHERE id=%s AND deleted_at IS NULL
+                RETURNING *
+                """,
+                (
+                    regeneration_prepared,
+                    regeneration_prepared,
+                    regeneration_prepared,
+                    regeneration_prepared,
+                    regeneration_prepared,
+                    regeneration_prepared,
+                    regeneration_prepared,
+                    regeneration_prepared,
+                    preview_run_id,
+                    len(selected_ids),
+                    now,
+                    task_id,
+                ),
+            ).fetchone()
+        return _public_task(_decode_row(row)) or {}, selected_ids
+
+    def mark_preview_running(self, task_id: str, preview_run_id: str) -> bool:
+        with self.connect() as conn:
+            row = conn.execute(
+                """
+                UPDATE data_process_tasks
+                SET preview_status='running', updated_at=%s
+                WHERE id=%s AND deleted_at IS NULL
+                  AND preview_status='queued' AND preview_run_id=%s
+                RETURNING id
+                """,
+                (utcnow(), task_id, preview_run_id),
+            ).fetchone()
+        return row is not None
+
+    def preview_is_running(self, task_id: str, preview_run_id: str) -> bool:
+        with self.connect() as conn:
+            row = conn.execute(
+                """
+                SELECT preview_status, preview_run_id
+                FROM data_process_tasks
+                WHERE id=%s AND deleted_at IS NULL
+                """,
+                (task_id,),
+            ).fetchone()
+        return bool(
+            row
+            and row.get("preview_status") in ACTIVE_PREVIEW_STATUSES
+            and row.get("preview_run_id") == preview_run_id
+        )
+
+    def update_preview_progress(
+        self,
+        task_id: str,
+        preview_run_id: str,
+        completed_files: int,
+        total_files: int,
+    ) -> bool:
+        total = max(1, total_files)
+        completed = min(max(0, completed_files), total)
+        progress = completed / total * 100
+        with self.connect() as conn:
+            row = conn.execute(
+                """
+                UPDATE data_process_tasks
+                SET preview_progress=%s, preview_completed_files=%s, updated_at=%s
+                WHERE id=%s AND deleted_at IS NULL
+                  AND preview_status='running' AND preview_run_id=%s
+                RETURNING id
+                """,
+                (progress, completed, utcnow(), task_id, preview_run_id),
+            ).fetchone()
+        return row is not None
+
+    def complete_preview(self, task_id: str, preview_run_id: str) -> bool:
+        with self.connect() as conn:
+            row = conn.execute(
+                """
+                UPDATE data_process_tasks
+                SET workflow_step='preview', preview_status='completed',
+                    preview_progress=100, preview_run_id=NULL,
+                    preview_failure_reason=NULL,
+                    preview_completed_files=preview_total_files, updated_at=%s
+                WHERE id=%s AND deleted_at IS NULL
+                  AND preview_status='running' AND preview_run_id=%s
+                RETURNING id
+                """,
+                (utcnow(), task_id, preview_run_id),
+            ).fetchone()
+        return row is not None
+
+    def mark_preview_failed(
+        self,
+        task_id: str,
+        reason: str,
+        *,
+        preview_run_id: str,
+    ) -> bool:
+        with self.connect() as conn:
+            row = conn.execute(
+                """
+                UPDATE data_process_tasks
+                SET preview_status='failed', preview_run_id=NULL,
+                    preview_failure_reason=%s, updated_at=%s
+                WHERE id=%s AND deleted_at IS NULL
+                  AND preview_status IN ('queued', 'running') AND preview_run_id=%s
+                RETURNING id
+                """,
+                (reason[:4000], utcnow(), task_id, preview_run_id),
+            ).fetchone()
+        return row is not None
+
+    def preview_progress(self, task_id: str) -> dict[str, Any]:
+        task = self.get_task(task_id)
+        return {
+            "task_id": task["id"],
+            "workflow_step": task.get("workflow_step") or "create",
+            "preview_status": task.get("preview_status") or "idle",
+            "preview_progress": float(task.get("preview_progress") or 0),
+            "preview_run_id": task.get("preview_run_id"),
+            "preview_failure_reason": task.get("preview_failure_reason"),
+            "preview_total_files": int(task.get("preview_total_files") or 0),
+            "preview_completed_files": int(task.get("preview_completed_files") or 0),
+        }
+
     def list_preview_items(
         self,
         task_id: str,
@@ -1332,7 +1625,7 @@ class DataProcessStore:
             UPDATE data_process_tasks
             SET status='pending', progress=20, output_count=0, filtered_count=0,
                 duplicate_count=0, error_count=0, failure_reason=NULL,
-                generation_run_id=NULL, updated_at=%s
+                generation_run_id=NULL, results_confirmed=FALSE, updated_at=%s
             WHERE id=%s
             """,
             (now, task_id),
@@ -1348,6 +1641,8 @@ class DataProcessStore:
                 raise InvalidStateError("published task cannot be regenerated")
             if task["status"] == "running":
                 raise ConflictError("data process task is already running")
+            if task.get("preview_status") in ACTIVE_PREVIEW_STATUSES:
+                raise ConflictError("preview is still running")
             preview_count = conn.execute(
                 "SELECT COUNT(*) AS count FROM data_process_preview_items WHERE task_id=%s",
                 (task_id,),
@@ -1365,7 +1660,8 @@ class DataProcessStore:
                 SET config=%s, status='running', progress=30, failure_reason=NULL,
                     started_at=%s, completed_at=NULL, output_dataset_id=NULL,
                     output_count=0, filtered_count=0, duplicate_count=0, error_count=0,
-                    generation_run_id=%s, updated_at=%s
+                    generation_run_id=%s, results_confirmed=FALSE,
+                    workflow_step='generate', updated_at=%s
                 WHERE id=%s
                 RETURNING *
                 """,
@@ -1391,10 +1687,19 @@ class DataProcessStore:
         return _decode_row(row) or {}
 
     def generation_is_running(self, task_id: str, generation_run_id: str) -> bool:
-        task = self.get_task(task_id)
-        return (
-            task["status"] == "running"
-            and task.get("generation_run_id") == generation_run_id
+        with self.connect() as conn:
+            row = conn.execute(
+                """
+                SELECT status, generation_run_id
+                FROM data_process_tasks
+                WHERE id=%s AND deleted_at IS NULL
+                """,
+                (task_id,),
+            ).fetchone()
+        return bool(
+            row
+            and row.get("status") == "running"
+            and row.get("generation_run_id") == generation_run_id
         )
 
     def update_generation_progress(
@@ -1469,7 +1774,8 @@ class DataProcessStore:
                 UPDATE data_process_tasks
                 SET status='completed', progress=100, output_count=%s, filtered_count=%s,
                     duplicate_count=%s, error_count=%s, failure_reason=NULL,
-                    completed_at=%s, generation_run_id=NULL, updated_at=%s
+                    completed_at=%s, generation_run_id=NULL, results_confirmed=FALSE,
+                    updated_at=%s
                 WHERE id=%s AND generation_run_id=%s RETURNING *
                 """,
                 (
@@ -1519,10 +1825,59 @@ class DataProcessStore:
             "duplicate_count": int(task.get("duplicate_count") or 0),
             "error_count": int(task.get("error_count") or 0),
             "failure_reason": task.get("failure_reason"),
+            "results_confirmed": bool(task.get("results_confirmed")),
             "started_at": task.get("started_at"),
             "completed_at": task.get("completed_at"),
         }
 
+    def confirm_results(self, task_id: str) -> dict[str, Any]:
+        """确认第六步结果，确认前再次校验所有生成记录。"""
+
+        with self.connect() as conn:
+            task = self._task_in_connection(conn, task_id, for_update=True)
+            if task["status"] != "completed":
+                raise InvalidStateError("only a completed task can confirm results")
+            if task.get("workflow_step") != "results":
+                raise InvalidStateError("workflow must be on results before confirmation")
+            if task.get("results_confirmed"):
+                return task
+
+            rows = conn.execute(
+                """
+                SELECT status, instruction, output
+                FROM data_process_results
+                WHERE task_id=%s
+                """,
+                (task_id,),
+            ).fetchall()
+            if not rows:
+                raise InvalidStateError("task has no results to confirm")
+            invalid_count = sum(
+                1
+                for row in rows
+                if row["status"] == "invalid"
+                or not str(row.get("instruction") or "").strip()
+                or not str(row.get("output") or "").strip()
+                or (
+                    _task_output_type(task) == "reasoning"
+                    and not _reasoning_output_is_valid(row.get("output"))
+                )
+            )
+            if invalid_count:
+                raise InvalidStateError(
+                    f"task contains {invalid_count} invalid results"
+                )
+
+            row = conn.execute(
+                """
+                UPDATE data_process_tasks
+                SET results_confirmed=TRUE, updated_at=%s
+                WHERE id=%s RETURNING *
+                """,
+                (utcnow(), task_id),
+            ).fetchone()
+        return _decode_row(row) or {}
+
     def list_results(
         self,
         task_id: str,
@@ -1651,6 +2006,83 @@ class DataProcessStore:
             )
         return _decode_row(row) or {}
 
+    def replace_generated_result(
+        self,
+        task_id: str,
+        result_id: str,
+        replacement: dict[str, Any],
+        *,
+        expected_updated_at: str,
+    ) -> dict[str, Any]:
+        """用新模型结果原位替换失败项，并将新内容设为恢复基线。"""
+
+        with self.connect() as conn:
+            task = self._task_in_connection(conn, task_id, for_update=True)
+            if task["status"] != "completed" or task.get("workflow_step") != "results":
+                raise InvalidStateError("task is not editing generation results")
+            if task.get("results_confirmed"):
+                raise InvalidStateError("confirmed results cannot be regenerated")
+            if task.get("output_dataset_id"):
+                raise InvalidStateError("published results cannot be regenerated")
+
+            current = conn.execute(
+                """SELECT * FROM data_process_results
+                   WHERE id=%s AND task_id=%s FOR UPDATE""",
+                (result_id, task_id),
+            ).fetchone()
+            if not current:
+                raise NotFoundError("data process result not found")
+            if current.get("status") != "invalid":
+                raise InvalidStateError("only an invalid result can be regenerated")
+            current_updated_at = _serialize_value(current.get("updated_at"))
+            if expected_updated_at != current_updated_at:
+                raise ConflictError("data process result was modified by another request")
+
+            instruction = str(replacement.get("instruction") or "").strip()
+            input_text = str(replacement.get("input") or "").strip()
+            output = str(replacement.get("output") or "").strip()
+            quality_score = replacement.get("quality_score") or {}
+            if not instruction or not output or not bool(quality_score.get("is_valid")):
+                raise InvalidStateError("regenerated result did not pass quality validation")
+            if _task_output_type(task) == "reasoning" and not _reasoning_output_is_valid(output):
+                raise InvalidStateError("regenerated reasoning result has an invalid output format")
+
+            now = utcnow()
+            row = conn.execute(
+                """
+                UPDATE data_process_results
+                SET instruction=%s, input=%s, output=%s,
+                    original_instruction=%s, original_input=%s, original_output=%s,
+                    status='valid', error=NULL, quality_score=%s, updated_at=%s
+                WHERE id=%s AND task_id=%s
+                RETURNING *
+                """,
+                (
+                    instruction,
+                    input_text,
+                    output,
+                    instruction,
+                    input_text,
+                    output,
+                    json_dumps(quality_score),
+                    now,
+                    result_id,
+                    task_id,
+                ),
+            ).fetchone()
+            conn.execute(
+                """
+                UPDATE data_process_tasks
+                SET error_count=(
+                    SELECT COUNT(*) FROM data_process_results
+                    WHERE task_id=%s AND status='invalid'
+                ), updated_at=%s
+                WHERE id=%s
+                """,
+                (task_id, now, task_id),
+            )
+        return _decode_row(row) or {}
+
     def get_generation_model(self, model_id: str) -> dict[str, Any]:
         with self.connect() as conn:
             row = conn.execute(
@@ -1704,6 +2136,8 @@ class DataProcessStore:
                 )
             if task["status"] != "completed":
                 raise InvalidStateError("only a completed task can be published")
+            if not task.get("results_confirmed"):
+                raise InvalidStateError("results must be confirmed before publishing")
             rows = conn.execute(
                 """
                 SELECT * FROM data_process_results