From 396d3f6f470a1c00e3019e8496ac56e96b43b167 Mon Sep 17 00:00:00 2001 From: caoxiaozhu Date: Sat, 25 Jul 2026 22:40:55 +0800 Subject: [PATCH] =?UTF-8?q?feat(data-process):=20=E6=94=AF=E6=8C=81?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E9=87=8D=E6=96=B0=E7=94=9F=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/v1/endpoints/data_process.py | 14 + backend/app/modules/data_process/store.py | 176 +++++++++++- backend/app/schemas/data_process.py | 30 ++ backend/tests/test_data_process_api.py | 163 ++++++++++- backend/tests/test_data_process_store.py | 288 +++++++++++++++++++ 5 files changed, 667 insertions(+), 4 deletions(-) diff --git a/backend/app/api/v1/endpoints/data_process.py b/backend/app/api/v1/endpoints/data_process.py index 1d1d0cd..afd8d05 100644 --- a/backend/app/api/v1/endpoints/data_process.py +++ b/backend/app/api/v1/endpoints/data_process.py @@ -68,6 +68,7 @@ from app.modules.data_process.store import ( new_id, ) from app.schemas.data_process import ( + DataProcessRegenerateRequest, DataProcessStatus, DataProcessTaskCreate, DataProcessTaskUpdate, @@ -732,6 +733,19 @@ def update_task( ) +@router.post("/{task_id}/regenerate") +def prepare_regeneration( + task_id: str, + payload: DataProcessRegenerateRequest, + store: DataProcessStore = Depends(get_data_process_store), +) -> dict[str, Any]: + with api_errors(): + return ok( + store.prepare_regeneration(task_id, payload.model_dump(mode="json")), + "data process task prepared for regeneration", + ) + + @router.delete("/{task_id}") def delete_task( task_id: str, diff --git a/backend/app/modules/data_process/store.py b/backend/app/modules/data_process/store.py index 09c0de2..6765460 100644 --- a/backend/app/modules/data_process/store.py +++ b/backend/app/modules/data_process/store.py @@ -20,6 +20,28 @@ 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"} +_PREVIEW_CONFIG_ALIASES = { + "preprocess_options": "preprocessOptions", + "chunk_method": "chunkMethod", + "chunk_size": "chunkSize", + "chunk_overlap": "chunkOverlap", + "min_chunk_size": "minChunkSize", + "semantic_breakpoint_percentile": "semanticBreakpointPercentile", + "preserve_tables": "preserveTables", + "preserve_code_blocks": "preserveCodeBlocks", + "preserve_lists": "preserveLists", +} +_UNSTRUCTURED_PREVIEW_DEFAULTS: dict[str, Any] = { + "chunk_method": "layout_hybrid", + "chunk_size": 800, + "chunk_overlap": 100, + "min_chunk_size": 100, + "semantic_breakpoint_percentile": 95, + "preserve_tables": True, + "preserve_code_blocks": True, + "preserve_lists": True, +} + class DataProcessStoreError(RuntimeError): pass @@ -64,6 +86,49 @@ def _json_value(value: Any, default: Any) -> Any: return default +def _preview_config_value(config: dict[str, Any], key: str, default: Any) -> Any: + if key in config: + return config[key] + return config.get(_PREVIEW_CONFIG_ALIASES[key], default) + + +def _normalized_preprocess_options(config: dict[str, Any]) -> Any: + value = _preview_config_value(config, "preprocess_options", []) + if isinstance(value, (list, tuple, set)): + return tuple(sorted({str(item) for item in value})) + return value + + +def _preview_config_projection( + process_type: str, + config: dict[str, Any], +) -> dict[str, Any]: + """只投影会改变预览切片的配置。 + + 生成模型、提示词、温度等参数不影响源文切片，因此不应该 + 破坏用户已经校对过的预览内容。 + """ + + projection: dict[str, Any] = { + "preprocess_options": _normalized_preprocess_options(config), + } + if process_type != "unstructured": + return projection + for key, default in _UNSTRUCTURED_PREVIEW_DEFAULTS.items(): + projection[key] = _preview_config_value(config, key, default) + return projection + + +def _preview_config_changed( + process_type: str, + current_config: dict[str, Any], + next_config: dict[str, Any], +) -> bool: + return _preview_config_projection(process_type, current_config) != _preview_config_projection( + process_type, next_config + ) + + def _serialize_value(value: Any) -> Any: if isinstance(value, (datetime, date)): return value.isoformat().replace("+00:00", "Z") @@ -361,6 +426,102 @@ class DataProcessStore: raise ConflictError("data process task name already exists") from exc return _decode_row(row) or {} + def prepare_regeneration( + self, + task_id: str, + payload: dict[str, Any], + ) -> dict[str, Any]: + """在一个事务中将已有任务恢复为可重新生成状态。 + + 已发布数据集是可被其他任务使用的独立产物，此处只解除当前任务的 + 输出指针，不删除数据集实体。用户完成新一轮生成后，可再次发布 + 以原子替换三个分割数据集的内容。 + """ + + try: + with self.connect() as conn: + task = self._task_in_connection(conn, task_id, for_update=True) + if task["status"] == "running": + raise ConflictError("running task cannot be prepared for regeneration") + + current_updated_at = _serialize_value(task.get("updated_at")) + if payload["expected_updated_at"] != current_updated_at: + raise ConflictError("data process task was modified by another request") + + process_type = str(payload["process_type"]) + if process_type != str(task["process_type"]): + raise InvalidStateError("process_type cannot be changed during regeneration") + + current_config = dict(task.get("config") or {}) + next_config = dict(payload.get("config") or {}) + preview_invalidated = _preview_config_changed( + process_type, + current_config, + next_config, + ) + published_row = conn.execute( + """ + SELECT EXISTS( + SELECT 1 FROM datasets + WHERE source_task_id=%s AND source='task' + ) AS exists + """, + (task_id,), + ).fetchone() + published_outputs_preserved = bool(task.get("output_dataset_id")) or bool( + published_row and published_row.get("exists") + ) + + # 删除顺序保证结果不再指向即将被替换的切片。 + conn.execute( + "DELETE FROM data_process_results WHERE task_id=%s", + (task_id,), + ) + if preview_invalidated: + conn.execute( + "DELETE FROM data_process_preview_items WHERE task_id=%s", + (task_id,), + ) + preview_count = 0 + else: + preview_row = conn.execute( + """ + SELECT COUNT(*) AS count FROM data_process_preview_items + WHERE task_id=%s + """, + (task_id,), + ).fetchone() + preview_count = int((preview_row or {}).get("count") or 0) + + now = utcnow() + row = conn.execute( + """ + UPDATE data_process_tasks + SET name=%s, description=%s, config=%s, status='pending', progress=%s, + output_dataset_id=NULL, 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, + updated_at=%s + WHERE id=%s + RETURNING * + """, + ( + payload["name"], + payload.get("description") or "", + json_dumps(next_config), + 20 if preview_count else 0, + now, + task_id, + ), + ).fetchone() + except psycopg.errors.UniqueViolation as exc: + raise ConflictError("data process task name already exists") from exc + return { + "task": _decode_row(row) or {}, + "preview_invalidated": preview_invalidated, + "published_outputs_preserved": published_outputs_preserved, + } + 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) @@ -480,12 +641,21 @@ class DataProcessStore: ), ).fetchone() created.append(_decode_row(row) or {}) + # 前端会只对本次新增的源文件构建预览，因此必须保留旧 + # 文件的切片。任何源文件增加都会使旧生成结果失效。 conn.execute("DELETE FROM data_process_results WHERE task_id=%s", (task_id,)) - conn.execute("DELETE FROM data_process_preview_items WHERE task_id=%s", (task_id,)) + preview_row = conn.execute( + """ + SELECT COUNT(*) AS count FROM data_process_preview_items + WHERE task_id=%s + """, + (task_id,), + ).fetchone() + preview_count = int((preview_row or {}).get("count") or 0) conn.execute( """ UPDATE data_process_tasks - SET status='pending', progress=0, output_count=0, filtered_count=0, + 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, input_count=( SELECT COALESCE(SUM(record_count), 0) @@ -494,7 +664,7 @@ class DataProcessStore: ), updated_at=%s WHERE id=%s """, - (task_id, now, task_id), + (20 if preview_count else 0, task_id, now, task_id), ) except psycopg.errors.UniqueViolation as exc: raise ConflictError( diff --git a/backend/app/schemas/data_process.py b/backend/app/schemas/data_process.py index e761b4b..ce2a03a 100644 --- a/backend/app/schemas/data_process.py +++ b/backend/app/schemas/data_process.py @@ -156,6 +156,36 @@ class DataProcessTaskUpdate(BaseModel): return self +class DataProcessRegenerateRequest(BaseModel): + """以一份完整配置准备任务重新生成。 + + ``expected_updated_at`` 用于防止详情页的旧快照覆盖其他人刚刚 + 保存的配置。重新生成不允许改变处理类型，避免旧源文件在新解析 + 规则下被静默误用。 + """ + + model_config = ConfigDict(extra="forbid") + + name: str = Field(min_length=1, max_length=150) + description: str + process_type: ProcessType + config: dict[str, Any] + expected_updated_at: str = Field(min_length=1) + + @field_validator("name") + @classmethod + def normalize_name(cls, value: str) -> str: + value = value.strip() + if not value: + raise ValueError("task name cannot be empty") + return value + + @model_validator(mode="after") + def validate_config(self) -> "DataProcessRegenerateRequest": + _validate_process_config(self.config) + return self + + class PreviewBuildRequest(BaseModel): model_config = ConfigDict(extra="forbid") diff --git a/backend/tests/test_data_process_api.py b/backend/tests/test_data_process_api.py index 905e97c..1c7ea3a 100644 --- a/backend/tests/test_data_process_api.py +++ b/backend/tests/test_data_process_api.py @@ -82,6 +82,43 @@ class FakeDataProcessStore: self.tasks[task_id].update(deepcopy(payload)) return self.get_task(task_id) + def prepare_regeneration(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]: + task = self.tasks.get(task_id) + if task is None: + raise NotFoundError("data process task not found") + if task["status"] == "running": + raise InvalidStateError("running task cannot be prepared for regeneration") + if payload["expected_updated_at"] != task.get("updated_at"): + raise InvalidStateError("data process task was modified by another request") + if payload["process_type"] != task["process_type"]: + raise InvalidStateError("process_type cannot be changed during regeneration") + published_outputs_preserved = bool(task.get("output_dataset_id")) + self.results[task_id] = [] + task.update( + { + "name": payload["name"], + "description": payload["description"], + "config": deepcopy(payload["config"]), + "status": "pending", + "progress": 20, + "output_dataset_id": None, + "output_count": 0, + "filtered_count": 0, + "duplicate_count": 0, + "error_count": 0, + "failure_reason": None, + "generation_run_id": None, + "started_at": None, + "completed_at": None, + "updated_at": "2026-07-25T20:00:00Z", + } + ) + return { + "task": deepcopy(task), + "preview_invalidated": False, + "published_outputs_preserved": published_outputs_preserved, + } + def delete_task(self, task_id: str, **_: Any) -> None: self.get_task(task_id) if self.tasks[task_id]["status"] == "running": @@ -133,7 +170,23 @@ class FakeDataProcessStore: if checksum in checksums or checksum in incoming: raise ValueError("the same source file content is already attached to this task") incoming.add(checksum) - return [self.add_source_file(task_id, **payload) for payload in files] + created = [self.add_source_file(task_id, **payload) for payload in files] + self.results[task_id] = [] + self.tasks[task_id].update( + { + "status": "pending", + "progress": 20 if self.previews[task_id] else 0, + "output_count": 0, + "filtered_count": 0, + "duplicate_count": 0, + "error_count": 0, + "failure_reason": None, + "generation_run_id": None, + "started_at": None, + "completed_at": None, + } + ) + return created def get_source_file( self, task_id: str, file_id: str, *, include_content: bool = True @@ -769,6 +822,81 @@ def test_external_source_never_returns_fake_success(tmp_path: Path) -> None: assert response.json()["detail"]["code"] == 501 +def test_regenerate_endpoint_prepares_an_existing_published_task(tmp_path: Path) -> None: + client, store, _ = make_client(tmp_path) + task_id = client.post( + "/modelTF/data-process", + json={"name": "待重新生成", "process_type": "structured", "config": {}}, + ).json()["data"]["id"] + store.tasks[task_id].update( + { + "status": "completed", + "updated_at": "2026-07-25T19:00:00Z", + "output_dataset_id": "dataset_train", + "output_count": 2, + } + ) + store.previews[task_id] = [{"id": "preview_1", "edited_content": "原切片"}] + store.results[task_id] = [{"id": "result_1"}] + + response = client.post( + f"/modelTF/data-process/{task_id}/regenerate", + json={ + "name": "重新生成后名称", + "description": "更换生成模型", + "process_type": "structured", + "config": {"generation_model_id": "model_2"}, + "expected_updated_at": "2026-07-25T19:00:00Z", + }, + ) + + assert response.status_code == 200 + data = response.json()["data"] + assert data["task"]["status"] == "pending" + assert data["task"]["output_dataset_id"] is None + assert data["preview_invalidated"] is False + assert data["published_outputs_preserved"] is True + assert store.results[task_id] == [] + assert store.previews[task_id][0]["id"] == "preview_1" + + +def test_regenerate_endpoint_validates_snapshot_and_locked_process_type( + tmp_path: Path, +) -> None: + client, store, _ = make_client(tmp_path) + task_id = client.post( + "/modelTF/data-process", + json={"name": "并发校验", "process_type": "structured", "config": {}}, + ).json()["data"]["id"] + store.tasks[task_id]["updated_at"] = "2026-07-25T19:00:00Z" + payload = { + "name": "并发校验", + "description": "", + "process_type": "structured", + "config": {}, + "expected_updated_at": "stale", + } + + stale = client.post(f"/modelTF/data-process/{task_id}/regenerate", json=payload) + assert stale.status_code == 409 + + payload.update( + { + "process_type": "unstructured", + "expected_updated_at": "2026-07-25T19:00:00Z", + } + ) + locked_type = client.post(f"/modelTF/data-process/{task_id}/regenerate", json=payload) + assert locked_type.status_code == 409 + + for missing_field in ("expected_updated_at", "description", "config"): + missing_required_field = client.post( + f"/modelTF/data-process/{task_id}/regenerate", + json={key: value for key, value in payload.items() if key != missing_field}, + ) + assert missing_required_field.status_code == 422 + + def test_config_validation_and_stop_state(tmp_path: Path) -> None: client, store, _ = make_client(tmp_path) invalid = client.post( @@ -855,6 +983,39 @@ def test_upload_batch_is_atomic_and_empty_files_are_rejected(tmp_path: Path) -> assert _stored_files(storage) == [] +def test_incremental_upload_keeps_existing_file_previews(tmp_path: Path) -> None: + client, store, _ = make_client(tmp_path) + task_id = client.post( + "/modelTF/data-process", + json={"name": "增量上传预览", "process_type": "unstructured", "config": {}}, + ).json()["data"]["id"] + first_batch = client.post( + f"/modelTF/data-process/{task_id}/source-files", + files=[ + ("files", ("first.txt", "第一个文件内容".encode(), "text/plain")), + ("files", ("second.txt", "第二个文件内容".encode(), "text/plain")), + ], + ) + assert first_batch.status_code == 200 + first_source_ids = {item["id"] for item in first_batch.json()["data"]["files"]} + built = client.post(f"/modelTF/data-process/{task_id}/preview/build", json={}) + assert built.status_code == 200 + assert {item["source_file_id"] for item in built.json()["data"]["items"]} == first_source_ids + store.results[task_id] = [{"id": "old_result"}] + + third = client.post( + f"/modelTF/data-process/{task_id}/source-files", + files={"files": ("third.txt", "第三个文件内容".encode(), "text/plain")}, + ) + + assert third.status_code == 200 + previews = client.get(f"/modelTF/data-process/{task_id}/preview").json()["data"] + assert previews["total"] == 2 + assert {item["source_file_id"] for item in previews["items"]} == first_source_ids + assert store.results[task_id] == [] + assert store.tasks[task_id]["progress"] == 20 + + def test_upload_preserves_store_error_when_storage_rollback_fails( tmp_path: Path, monkeypatch: Any, diff --git a/backend/tests/test_data_process_store.py b/backend/tests/test_data_process_store.py index 604ae21..761faec 100644 --- a/backend/tests/test_data_process_store.py +++ b/backend/tests/test_data_process_store.py @@ -8,9 +8,12 @@ from typing import Any import pytest from app.modules.data_process.store import ( + ConflictError, DataProcessStore, DataProcessStoreError, + InvalidStateError, _decode_row, + _preview_config_changed, _source_storage_descriptor, ) @@ -114,6 +117,72 @@ class _PublishStore(DataProcessStore): return [] +class _RegenerationConnection: + def __init__(self, task: dict[str, Any]) -> None: + self.task = task + self.datasets = [ + {"id": "dataset_train"}, + {"id": "dataset_validation"}, + {"id": "dataset_test"}, + ] + self.sources = [{"id": "source_1"}] + self.previews = [{"id": "preview_1"}] + self.results = [{"id": "result_1"}] + + def execute(self, sql: str, params: Any = None) -> _Result: + normalized = " ".join(sql.split()) + if params is not None: + assert normalized.count("%s") == len(params) + if normalized.startswith("SELECT EXISTS("): + return _Result(row={"exists": bool(self.datasets)}) + if normalized.startswith("DELETE FROM data_process_results"): + self.results.clear() + return _Result() + if normalized.startswith("DELETE FROM data_process_preview_items"): + self.previews.clear() + return _Result() + if normalized.startswith("SELECT COUNT(*) AS count FROM data_process_preview_items"): + return _Result(row={"count": len(self.previews)}) + if normalized.startswith("UPDATE data_process_tasks SET name="): + self.task.update( + { + "name": params[0], + "description": params[1], + "config": params[2], + "status": "pending", + "progress": params[3], + "output_dataset_id": None, + "output_count": 0, + "filtered_count": 0, + "duplicate_count": 0, + "error_count": 0, + "failure_reason": None, + "generation_run_id": None, + "started_at": None, + "completed_at": None, + "updated_at": params[4], + } + ) + return _Result(row=dict(self.task)) + raise AssertionError(f"unexpected SQL: {normalized}") + + +class _RegenerationStore(DataProcessStore): + def __init__(self, conn: _RegenerationConnection): + self._conn = conn + + @contextmanager + def connect(self) -> Iterator[_RegenerationConnection]: + yield self._conn + + def _task_in_connection( + self, conn: Any, task_id: str, *, for_update: bool = False + ) -> dict[str, Any]: + assert for_update is True + assert task_id == self._conn.task["id"] + return dict(self._conn.task) + + def test_decode_row_serializes_postgres_numeric_values_as_json_numbers() -> None: decoded = _decode_row( { @@ -125,6 +194,225 @@ def test_decode_row_serializes_postgres_numeric_values_as_json_numbers() -> None assert decoded == {"progress": 100.0, "duration_seconds": 389.0} +@pytest.mark.parametrize( + ("process_type", "current", "next_config", "expected"), + [ + ( + "structured", + {"preprocess_options": ["deduplicate"]}, + {"preprocess_options": ["deduplicate"], "temperature": 0.2}, + False, + ), + ( + "structured", + {"preprocess_options": ["deduplicate"]}, + {"preprocess_options": ["clean_invalid"]}, + True, + ), + ( + "structured", + {"preprocess_options": ["a", "b"]}, + {"preprocessOptions": ["b", "a", "a"], "chunk_size": 2048}, + False, + ), + ( + "unstructured", + {"chunk_method": "fixed"}, + {"chunk_method": "fixed", "generation_prompt": "new"}, + False, + ), + ( + "unstructured", + {"chunk_method": "fixed"}, + {"chunk_method": "semantic"}, + True, + ), + ( + "unstructured", + {"chunk_size": 800, "chunk_overlap": 100}, + {"chunk_size": 900, "chunk_overlap": 100}, + True, + ), + ( + "unstructured", + {"min_chunk_size": 100}, + {"min_chunk_size": 120}, + True, + ), + ( + "unstructured", + {"semantic_breakpoint_percentile": 95}, + {"semantic_breakpoint_percentile": 90}, + True, + ), + ( + "unstructured", + {}, + { + "preserve_tables": True, + "preserve_code_blocks": True, + "preserve_lists": True, + }, + False, + ), + ( + "unstructured", + {"preserve_tables": False}, + {"preserve_tables": True}, + True, + ), + ( + "unstructured", + {"preserve_code_blocks": False}, + {"preserve_code_blocks": True}, + True, + ), + ( + "unstructured", + {"preserve_lists": False}, + {"preserve_lists": True}, + True, + ), + ( + "unstructured", + {"preprocess_options": ["deduplicate"]}, + {"preprocess_options": ["clean_invalid"]}, + True, + ), + ], +) +def test_regeneration_preview_invalidation_matrix( + process_type: str, + current: dict[str, Any], + next_config: dict[str, Any], + expected: bool, +) -> None: + assert _preview_config_changed(process_type, current, next_config) is expected + + +def _regeneration_task(**updates: Any) -> dict[str, Any]: + task = { + "id": "task-1", + "name": "原任务", + "description": "", + "process_type": "unstructured", + "config": {"chunk_method": "fixed", "temperature": 0.7}, + "status": "completed", + "progress": 100, + "output_dataset_id": "dataset_train", + "output_count": 28, + "filtered_count": 1, + "duplicate_count": 1, + "error_count": 0, + "failure_reason": None, + "generation_run_id": None, + "started_at": "2026-07-25T18:00:00Z", + "completed_at": "2026-07-25T18:05:00Z", + "updated_at": "2026-07-25T18:05:00Z", + } + task.update(updates) + return task + + +def test_prepare_regeneration_preserves_outputs_sources_and_generation_only_preview() -> None: + conn = _RegenerationConnection(_regeneration_task()) + original_datasets = list(conn.datasets) + original_sources = list(conn.sources) + + result = _RegenerationStore(conn).prepare_regeneration( + "task-1", + { + "name": "新任务名", + "description": "更换生成参数", + "process_type": "unstructured", + "config": {"chunk_method": "fixed", "temperature": 0.2}, + "expected_updated_at": "2026-07-25T18:05:00Z", + }, + ) + + assert result["preview_invalidated"] is False + assert result["published_outputs_preserved"] is True + assert result["task"]["output_dataset_id"] is None + assert result["task"]["status"] == "pending" + assert result["task"]["progress"] == 20 + assert result["task"]["output_count"] == 0 + assert result["task"]["started_at"] is None + assert result["task"]["completed_at"] is None + assert conn.results == [] + assert conn.previews == [{"id": "preview_1"}] + assert conn.datasets == original_datasets + assert conn.sources == original_sources + + +def test_prepare_regeneration_deletes_preview_when_chunk_configuration_changes() -> None: + conn = _RegenerationConnection(_regeneration_task(output_dataset_id=None)) + + result = _RegenerationStore(conn).prepare_regeneration( + "task-1", + { + "name": "原任务", + "description": "", + "process_type": "unstructured", + "config": {"chunk_method": "semantic", "temperature": 0.7}, + "expected_updated_at": "2026-07-25T18:05:00Z", + }, + ) + + assert result["preview_invalidated"] is True + assert result["published_outputs_preserved"] is True + assert result["task"]["progress"] == 0 + assert conn.previews == [] + assert conn.results == [] + assert len(conn.datasets) == 3 + + +@pytest.mark.parametrize( + ("task_updates", "payload_updates", "error_type", "message"), + [ + ( + {"status": "running"}, + {}, + ConflictError, + "running task cannot be prepared", + ), + ( + {}, + {"expected_updated_at": "2026-07-25T17:00:00Z"}, + ConflictError, + "modified by another request", + ), + ( + {}, + {"process_type": "structured"}, + InvalidStateError, + "process_type cannot be changed", + ), + ], +) +def test_prepare_regeneration_rejects_running_stale_and_type_change_without_mutation( + task_updates: dict[str, Any], + payload_updates: dict[str, Any], + error_type: type[Exception], + message: str, +) -> None: + conn = _RegenerationConnection(_regeneration_task(**task_updates)) + payload = { + "name": "原任务", + "description": "", + "process_type": "unstructured", + "config": {"chunk_method": "fixed"}, + "expected_updated_at": "2026-07-25T18:05:00Z", + **payload_updates, + } + + with pytest.raises(error_type, match=message): + _RegenerationStore(conn).prepare_regeneration("task-1", payload) + + assert conn.results == [{"id": "result_1"}] + assert conn.previews == [{"id": "preview_1"}] + assert conn.task["output_dataset_id"] == "dataset_train" + + def test_publish_creates_three_independent_datasets_with_exact_counts() -> None: results = [ {