fix(host-workspace): make config writes atomic and lock-protected

modules/host_workspace_manager.py

@@ -3,7 +3,10 @@
 from __future__ import annotations
 
 import json
+import os
 import re
+import tempfile
+import threading
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple, Union
 
@@ -11,6 +14,7 @@ from config import DEFAULT_PROJECT_PATH, HOST_WORKSPACES_FILE
 
 _WORKSPACE_ID_RE = re.compile(r"^[a-zA-Z0-9._-]{1,64}$")
 _REPO_ROOT = Path(__file__).resolve().parents[1]
+_HOST_WORKSPACE_LOCK = threading.RLock()
 
 
 def _resolve_config_path(config_path: Optional[Union[str, Path]] = None) -> Path:
@@ -43,17 +47,39 @@ def _default_payload() -> Dict[str, Any]:
     }
 
 
-def _ensure_config_file(path: Path) -> Dict[str, Any]:
+def _atomic_write_json(path: Path, data: Dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp_path = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent))
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as fp:
+            json.dump(data, fp, ensure_ascii=False, indent=2)
+            fp.flush()
+            os.fsync(fp.fileno())
+        os.replace(tmp_path, path)
+    finally:
+        try:
+            if os.path.exists(tmp_path):
+                os.remove(tmp_path)
+        except Exception:
+            pass
+
+
+def _ensure_config_file(path: Path, *, strict: bool = False) -> Dict[str, Any]:
     path.parent.mkdir(parents=True, exist_ok=True)
     if path.exists():
         try:
             data = json.loads(path.read_text(encoding="utf-8"))
             if isinstance(data, dict):
                 return data
-        except Exception:
-            pass
+            if strict:
+                raise RuntimeError("host_workspaces 配置格式错误（非 JSON 对象），已停止写入以避免覆盖原文件")
+        except Exception as exc:
+            if strict:
+                raise RuntimeError(f"host_workspaces 配置解析失败，已停止写入以避免覆盖原文件: {exc}") from exc
+            # 只回退到内存默认值，不覆盖磁盘文件
+            return _default_payload()
     data = _default_payload()
-    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+    _atomic_write_json(path, data)
     return data
 
 
@@ -107,7 +133,8 @@ def load_host_workspace_catalog(
     config_path: Optional[Union[str, Path]] = None,
 ) -> Dict[str, Any]:
     cfg_path = _resolve_config_path(config_path)
-    payload = _ensure_config_file(cfg_path)
+    with _HOST_WORKSPACE_LOCK:
+        payload = _ensure_config_file(cfg_path, strict=False)
     raw_workspaces = payload.get("workspaces")
     if not isinstance(raw_workspaces, list):
         raw_workspaces = []
@@ -174,56 +201,56 @@ def create_host_workspace(
     config_path: Optional[Union[str, Path]] = None,
 ) -> Dict[str, Any]:
     cfg_path = _resolve_config_path(config_path)
-    payload = _ensure_config_file(cfg_path)
+    with _HOST_WORKSPACE_LOCK:
+        payload = _ensure_config_file(cfg_path, strict=True)
 
-    raw_workspaces = payload.get("workspaces")
-    if not isinstance(raw_workspaces, list):
-        raw_workspaces = []
+        raw_workspaces = payload.get("workspaces")
+        if not isinstance(raw_workspaces, list):
+            raw_workspaces = []
 
-    normalized_path = _normalize_workspace_path(path)
-    normalized_path.mkdir(parents=True, exist_ok=True)
-    target_path_str = str(normalized_path)
+        normalized_path = _normalize_workspace_path(path)
+        normalized_path.mkdir(parents=True, exist_ok=True)
+        target_path_str = str(normalized_path)
 
-    for idx, item in enumerate(raw_workspaces):
-        existing = _normalize_entry(item, idx)
-        if not existing:
-            continue
-        if existing.get("path") == target_path_str:
-            # 已存在相同路径则直接返回
-            return {
-                "created": False,
-                "workspace": existing,
-                "catalog": load_host_workspace_catalog(config_path=cfg_path),
-            }
+        for idx, item in enumerate(raw_workspaces):
+            existing = _normalize_entry(item, idx)
+            if not existing:
+                continue
+            if existing.get("path") == target_path_str:
+                # 已存在相同路径则直接返回
+                return {
+                    "created": False,
+                    "workspace": existing,
+                    "catalog": load_host_workspace_catalog(config_path=cfg_path),
+                }
 
-    clean_label = str(label or "").strip()
-    base_id_seed = clean_label or normalized_path.name or "workspace"
-    base_id = _slugify_workspace_id(base_id_seed)
-    existing_ids = {
-        _normalize_workspace_id(item.get("workspace_id") or item.get("id"), i)
-        for i, item in enumerate(raw_workspaces)
-        if isinstance(item, dict)
-    }
-    workspace_id = base_id
-    suffix = 2
-    while workspace_id in existing_ids:
-        workspace_id = f"{base_id}-{suffix}"
-        suffix += 1
+        clean_label = str(label or "").strip()
+        base_id_seed = clean_label or normalized_path.name or "workspace"
+        base_id = _slugify_workspace_id(base_id_seed)
+        existing_ids = {
+            _normalize_workspace_id(item.get("workspace_id") or item.get("id"), i)
+            for i, item in enumerate(raw_workspaces)
+            if isinstance(item, dict)
+        }
+        workspace_id = base_id
+        suffix = 2
+        while workspace_id in existing_ids:
+            workspace_id = f"{base_id}-{suffix}"
+            suffix += 1
 
-    workspace = {
-        "workspace_id": workspace_id,
-        "label": clean_label or workspace_id,
-        "path": target_path_str,
-    }
-    raw_workspaces.append(workspace)
-    payload["workspaces"] = raw_workspaces
+        workspace = {
+            "workspace_id": workspace_id,
+            "label": clean_label or workspace_id,
+            "path": target_path_str,
+        }
+        raw_workspaces.append(workspace)
+        payload["workspaces"] = raw_workspaces
 
-    default_id = str(payload.get("default_workspace_id") or "").strip()
-    if set_default or not default_id:
-        payload["default_workspace_id"] = workspace_id
+        default_id = str(payload.get("default_workspace_id") or "").strip()
+        if set_default or not default_id:
+            payload["default_workspace_id"] = workspace_id
 
-    cfg_path.parent.mkdir(parents=True, exist_ok=True)
-    cfg_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+        _atomic_write_json(cfg_path, payload)
 
     return {
         "created": True,