diff --git a/studio/api/routers/projects/training.py b/studio/api/routers/projects/training.py index 12ce936..7ed73e5 100644 --- a/studio/api/routers/projects/training.py +++ b/studio/api/routers/projects/training.py @@ -78,7 +78,7 @@ from ....domain import RegAiConfig from ....infrastructure.event_bus import bus from ....paths import STUDIO_DATA, safe_join -from ....services import model_downloader, version_config +from ....services import model_downloader, task_snapshot, version_config from ....services import presets as preset_flow from ....services.tagging import caption_snapshot from ....services.reg import builder as reg_builder, dedup as reg_dedup @@ -963,6 +963,13 @@ def enqueue_version_training( str(cfg_path), req_tid, scheduled_at), ) tid = int(cur.lastrowid) + # 任务的模型和其他训练参数在入队这一刻就冻结。 + # 后续切换全局模型或编辑 version config 只影响新任务。 + frozen_cfg = task_snapshot.freeze_config(tid, cfg_path) + conn.execute( + "UPDATE tasks SET config_path = ? WHERE id = ?", + (str(frozen_cfg), tid), + ) conn.commit() # ADR-0007 PR-5: version.status 由 supervisor 在 _spawn_task 推到 training; # project 无 stage;这里不再 advance。 diff --git a/studio/api/routers/queue/io.py b/studio/api/routers/queue/io.py index df4485a..6159788 100644 --- a/studio/api/routers/queue/io.py +++ b/studio/api/routers/queue/io.py @@ -1,7 +1,7 @@ """Queue task snapshot(PR-6 commit 6 从 server.py 抽出)。 1 route: - GET /api/queue/{task_id}/snapshot/config ADR-0007 §11.7 task 启动 freeze 的 config + GET /api/queue/{task_id}/snapshot/config ADR-0007 §11.7 task 入队 freeze 的 config 历史:本文件曾有 /api/queue/export + /api/queue/import(预设池时代的队列 JSON 分享)。R-5 删除——现代任务 config 是 version 私有,导出恒空、导入恒跳过, @@ -22,7 +22,7 @@ @router.get("/api/queue/{task_id}/snapshot/config") def get_task_snapshot_config(task_id: int) -> dict[str, Any]: - """ADR-0007 §11.7:返回 task 启动时冻结的 config。 + """ADR-0007 §11.7:返回 task 入队时冻结的 config。 返回 ``{"yaml": str, "config": dict}``。task 不存在 / 无 snapshot → 404。 UI [关联配置] tab 用此 + 触发 "套用此配置" 路由跳转到 ⑦ 训练 phase + prefill。 diff --git a/studio/api/routers/queue/lifecycle.py b/studio/api/routers/queue/lifecycle.py index bdf6ed9..0c78b91 100644 --- a/studio/api/routers/queue/lifecycle.py +++ b/studio/api/routers/queue/lifecycle.py @@ -35,6 +35,7 @@ ) from ....infrastructure.event_bus import bus from ....paths import USER_PRESETS_DIR, task_dir +from ....services import task_snapshot from ....supervisor.resources import ( JOB_KIND_RESOURCE_CLASS, RESOURCE_EXCLUSIVE, @@ -216,7 +217,14 @@ def enqueue(body: EnqueueRequest) -> dict[str, Any]: task_id = db.create_task( conn, name=name, config_name=body.config_name, priority=body.priority, scheduled_at=body.scheduled_at, + commit=False, ) + frozen_cfg = task_snapshot.freeze_config(task_id, cfg_path) + conn.execute( + "UPDATE tasks SET config_path = ? WHERE id = ?", + (str(frozen_cfg), task_id), + ) + conn.commit() task = db.get_task(conn, task_id) bus.publish({ "type": "task_state_changed", @@ -476,6 +484,7 @@ def retry_task(task_id: int) -> dict[str, Any]: name=original["name"], config_name=original["config_name"], priority=original["priority"], + commit=False, ) copy_fields: dict[str, Any] = {} # R-3:task_type / params 必须一并复制——数据作业类 task 重跑靠它们 @@ -483,8 +492,18 @@ def retry_task(task_id: int) -> dict[str, Any]: for k in ("config_path", "project_id", "version_id", "task_type", "params"): if original.get(k) is not None: copy_fields[k] = original[k] + source_cfg = original.get("config_path") + if source_cfg and Path(source_cfg).exists(): + copy_fields["config_path"] = str( + task_snapshot.freeze_config(new_id, Path(source_cfg)) + ) if copy_fields: - db.update_task(conn, new_id, **copy_fields) + cols = ", ".join(f"{key} = ?" for key in copy_fields) + conn.execute( + f"UPDATE tasks SET {cols} WHERE id = ?", + [*copy_fields.values(), new_id], + ) + conn.commit() new_task = db.get_task(conn, new_id) bus.publish( {"type": "task_state_changed", "task_id": new_id, "status": "pending"} diff --git a/studio/infrastructure/db.py b/studio/infrastructure/db.py index 3a63ab0..00fe0ab 100644 --- a/studio/infrastructure/db.py +++ b/studio/infrastructure/db.py @@ -136,9 +136,13 @@ def create_task( params: Optional[dict[str, Any]] = None, project_id: Optional[int] = None, version_id: Optional[int] = None, + commit: bool = True, ) -> int: """建 pending(或 scheduled)task。 + ``commit=False`` 仅供需要在同一事务内先写入 task-scoped + config 快照的入队端点使用;其他调用保持原有的立即提交行为。 + R-2 台账合并:tasks 表承接全部工作项。`task_type` 缺省 'train'(老调用方 兼容);数据作业类(download/tag/…)带 `params`(kind 专属参数 JSON)+ project_id/version_id 入库。写路径切换(services 从 create_job 改到这里) @@ -165,7 +169,8 @@ def create_task( _json.dumps(params) if params is not None else None, project_id, version_id), ) - conn.commit() + if commit: + conn.commit() return int(cur.lastrowid) diff --git a/studio/infrastructure/paths.py b/studio/infrastructure/paths.py index 087f066..1662b29 100644 --- a/studio/infrastructure/paths.py +++ b/studio/infrastructure/paths.py @@ -77,7 +77,7 @@ def resolve_studio_data(pointer_file: Path | None = None) -> Path: # Task-scoped 档案根目录。每个 task 独立子目录,跟 version 解耦, # 删 version 不会带走 task 历史(loss / 参数 / sample / 日志)。 # 子目录约定(snapshot/ 已由 task_snapshot.py 引入 ADR-0007 §11.7): -# tasks//snapshot/config.yaml ← task 启动时 freeze 的 config +# tasks//snapshot/config.yaml ← task 入队时 freeze 的 config # tasks//monitor/state.json ← 训练监控状态(loss/LR/sample 索引) # tasks//samples/*.png ← 训练采样图 # tasks//run.log ← worker 子进程 stdout/stderr diff --git a/studio/services/task_snapshot.py b/studio/services/task_snapshot.py index 71a5f0f..668632c 100644 --- a/studio/services/task_snapshot.py +++ b/studio/services/task_snapshot.py @@ -1,14 +1,14 @@ """Task config snapshot — ADR-0007 §11.7。 -task 启动时把当时的 version config.yaml 冻结一份到 +task 入队时把当时的 version config.yaml 冻结一份到 ``studio_data/tasks/{task_id}/snapshot/config.yaml``。 设计要点: - **仅冻 config**,不冻 caption / 图 / 正则集(跨 OS export OK,磁盘代价 KB 级) - 心智分离 UI:task 详情独立 [关联配置] tab,**不点 task 跳 version config 编辑页** → 让 user 理解 config 是历史快照,caption / 图是 version 当前状态 -- 冻结时机:supervisor `_spawn_task` 把 cfg_path 给 worker 之前 -- 失败不阻塞 task 启动(snapshot 是 forensics 不是必需) +- 冻结时机:新 task 入队时;老 task 仍由 supervisor 启动时补冻结 +- 新 task 若无法冻结则不入队;老 task 的启动时补冻结仍为非阻塞 用 user 视角:"点 task 详情 [关联配置] 看当时跑的什么参数,按'套用此配置'按钮 跳到 ⑦ 训练 phase 页面 + prefill → 编辑 → 训练 = 新 task" (§11.7 流程)。 @@ -53,6 +53,11 @@ def freeze_config(task_id: int, source: Path) -> Path: raise FileNotFoundError(f"snapshot source not found: {source}") dst = snapshot_config_path(task_id) dst.parent.mkdir(parents=True, exist_ok=True) + # 新 task 入队时已把 config_path 指向这份快照。supervisor + # 启动时仍会调用本函数以兼容老 task,此时 source == dst, + # 应直接复用,不要让 shutil.copy2 报 SameFileError。 + if dst.exists() and source.samefile(dst): + return dst shutil.copy2(source, dst) return dst diff --git a/studio/supervisor/core.py b/studio/supervisor/core.py index 43b340f..6844cae 100644 --- a/studio/supervisor/core.py +++ b/studio/supervisor/core.py @@ -710,9 +710,10 @@ def _fail_task_config_missing( ) def _freeze_task_snapshot(self, task_id: int, cfg_path: Path) -> None: - """ADR-0007 §11.7 / PR-3 commit 4:task 启动 → 冻结当时的 config - 到 studio_data/tasks/{tid}/snapshot/config.yaml。失败不阻 task - 启动(snapshot 是 forensics 不是必需)。 + """ADR-0007 §11.7:兼容老 task,启动时补冻结 config。 + + 新 task 在入队时已冻结,此处为同一文件的幂等调用。补冻结失败 + 不阻塞老 task 启动。 """ try: from ..services import task_snapshot diff --git a/tests/test_studio_queue_endpoints.py b/tests/test_studio_queue_endpoints.py index 778a5d4..069f1d7 100644 --- a/tests/test_studio_queue_endpoints.py +++ b/tests/test_studio_queue_endpoints.py @@ -23,6 +23,7 @@ def isolated(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: """ from studio.api.routers import logs as _logs_router from studio.api.routers.queue import lifecycle as _queue_lifecycle + from studio.infrastructure import paths as _paths dbfile = tmp_path / "studio.db" db.init_db(dbfile) @@ -39,6 +40,7 @@ def isolated(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: monkeypatch.setattr(server.db, "STUDIO_DB", dbfile) # connect() 默认路径 # PR-6 commit 6:queue lifecycle 用自己 import 的 USER_PRESETS_DIR monkeypatch.setattr(_queue_lifecycle, "USER_PRESETS_DIR", presets) + monkeypatch.setattr(_paths, "TASKS_DIR", tmp_path / "tasks") # PR-6 commit 1:logs router 用自己 import 的 LOGS_DIR monkeypatch.setattr(_logs_router, "LOGS_DIR", logs) return tmp_path @@ -87,6 +89,9 @@ def test_enqueue_and_get(client: TestClient) -> None: assert data["config_name"] == "good" assert data["status"] == "pending" tid = data["id"] + from studio.services import task_snapshot + assert data["config_path"] == str(task_snapshot.snapshot_config_path(tid)) + assert task_snapshot.snapshot_config_path(tid).exists() got = client.get(f"/api/queue/{tid}") assert got.status_code == 200 @@ -170,6 +175,19 @@ def test_retry_copies_full_training_context(client: TestClient) -> None: assert new.get("monitor_state_path") is None +def test_retry_gets_its_own_frozen_config(client: TestClient) -> None: + from studio.services import task_snapshot + + tid = client.post("/api/queue", json={"config_name": "good"}).json()["id"] + with db.connection_for() as conn: + db.update_task(conn, tid, status="failed") + + new = client.post(f"/api/queue/{tid}/retry").json() + expected = task_snapshot.snapshot_config_path(new["id"]) + assert new["config_path"] == str(expected) + assert expected.read_text(encoding="utf-8") == "epochs: 1\n" + + def test_outputs_list_with_files( client: TestClient, isolated, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/test_task_snapshot.py b/tests/test_task_snapshot.py index be91f8b..d458e70 100644 --- a/tests/test_task_snapshot.py +++ b/tests/test_task_snapshot.py @@ -60,6 +60,17 @@ def test_freeze_config_overwrites_existing(isolated, tmp_path: Path) -> None: assert final["config"]["lr"] == 0.0005 +def test_freeze_config_is_idempotent_when_source_is_snapshot( + isolated, tmp_path: Path +) -> None: + src = tmp_path / "cfg.yaml" + src.write_text("lr: 0.001\n", encoding="utf-8") + dst = task_snapshot.freeze_config(7, src) + + assert task_snapshot.freeze_config(7, dst) == dst + assert dst.read_text(encoding="utf-8") == "lr: 0.001\n" + + def test_freeze_config_missing_source_raises(isolated, tmp_path: Path) -> None: with pytest.raises(FileNotFoundError): task_snapshot.freeze_config(7, tmp_path / "nonexistent.yaml") diff --git a/tests/test_train_endpoints.py b/tests/test_train_endpoints.py index 08d4940..5698ba7 100644 --- a/tests/test_train_endpoints.py +++ b/tests/test_train_endpoints.py @@ -13,11 +13,13 @@ @pytest.fixture def env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + from studio.infrastructure import paths as _paths dbfile = tmp_path / "studio.db" db.init_db(dbfile) monkeypatch.setattr(db, "STUDIO_DB", dbfile) monkeypatch.setattr(server.db, "STUDIO_DB", dbfile) monkeypatch.setattr(projects, "PROJECTS_DIR", tmp_path / "projects") + monkeypatch.setattr(_paths, "TASKS_DIR", tmp_path / "tasks") presets_dir = tmp_path / "presets" presets_dir.mkdir() from studio.services.presets import io as presets_io @@ -315,10 +317,55 @@ def test_enqueue_creates_task_with_ids_and_config_path( assert task["status"] == "pending" assert task["project_id"] == pid assert task["version_id"] == vid - assert task["config_path"] and task["config_path"].endswith("config.yaml") + from studio.services import task_snapshot + expected = task_snapshot.snapshot_config_path(task["id"]) + assert task["config_path"] == str(expected) + assert expected.exists() # ADR-0007 PR-5: version.status 由 supervisor 在 spawn 时推 training;enqueue 时仍 preparing +def test_each_enqueued_task_keeps_model_selected_at_that_moment( + client: TestClient, env, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +) -> None: + """切换全局模型后,旧 task 仍用入队时的模型,新 task 用新模型。""" + from studio.services import task_snapshot, version_config + + pid, vid = _make(client) + _seed_preset(env, "tpl") + client.post( + f"/api/projects/{pid}/versions/{vid}/config/from_preset", + json={"name": "tpl"}, + ) + + selected = {"path": (tmp_path / "model-a.safetensors").as_posix()} + + def overlay(data): + result = dict(data) + result["transformer_path"] = selected["path"] + return result + + monkeypatch.setattr(version_config, "apply_global_path_overlay", overlay) + + first = client.post(f"/api/projects/{pid}/versions/{vid}/queue").json() + first_snapshot = task_snapshot.read_snapshot_config(first["id"]) + assert first_snapshot is not None + assert first_snapshot["config"]["transformer_path"] == selected["path"] + + with db.connection_for(env["db"]) as conn: + db.update_task(conn, first["id"], status="done") + + selected["path"] = (tmp_path / "model-b.safetensors").as_posix() + second = client.post(f"/api/projects/{pid}/versions/{vid}/queue").json() + second_snapshot = task_snapshot.read_snapshot_config(second["id"]) + assert second_snapshot is not None + assert second_snapshot["config"]["transformer_path"] == selected["path"] + + # Первая задача не изменилась после переключения. + first_snapshot = task_snapshot.read_snapshot_config(first["id"]) + assert first_snapshot is not None + assert first_snapshot["config"]["transformer_path"].endswith("model-a.safetensors") + + def test_enqueue_rejects_active_task(client: TestClient, env) -> None: """同 version 已有 pending/running task → 409。""" pid, vid = _make(client)