fix(queue): freeze training model when task is enqueued

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。

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。

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,15 +484,26 @@ 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 重跑靠它们
         # 路由 worker 与还原参数（漏了会退化成 train 跑错脚本）。
         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"}

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)
 
 

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/<id>/snapshot/config.yaml   ← task 启动时 freeze 的 config
+#   tasks/<id>/snapshot/config.yaml   ← task 入队时 freeze 的 config
 #   tasks/<id>/monitor/state.json     ← 训练监控状态（loss/LR/sample 索引）
 #   tasks/<id>/samples/*.png          ← 训练采样图
 #   tasks/<id>/run.log                ← worker 子进程 stdout/stderr

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
 

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

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:

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")

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)