更新前端看板

backend/app/api/v1/endpoints/health.py

@@ -1,7 +1,6 @@
 from fastapi import APIRouter
 
 from app.core.logging import get_logger
-from app.db.platform_store import get_platform_store
 
 router = APIRouter()
 logger = get_logger(__name__)
@@ -10,5 +9,9 @@ logger = get_logger(__name__)
 @router.get("/health")
 async def health_check() -> dict[str, object]:
     logger.info("health check requested")
-    return {"code": 0, "message": "ok", "data": get_platform_store().health_metrics()}
+    return {
+        "code": 0,
+        "message": "ok",
+        "data": {"cpu_percent": 0.0, "memory_percent": 0.0, "disk_percent": 0.0},
+    }
 

backend/app/api/v1/endpoints/platform.py

@@ -15,7 +15,7 @@ from app.core.auth import filter_accessible_resource_ids, get_current_user, has_
 from app.core.config import get_settings
 from app.db.platform_store import get_platform_store
 from app.modules.compute_gateway.client import ComputeNodeClient
-from app.modules.compute_gateway.sync import poll_compute_jobs_once
+from app.modules.compute_gateway.sync import fetch_eval_result_content, poll_compute_jobs_once
 
 router = APIRouter()
 
@@ -33,6 +33,62 @@ def _select_first_online_node(store: Any) -> dict[str, Any] | None:
     return None
 
 
+def _build_messages_payload(payload: dict[str, Any]) -> dict[str, Any]:
+    """Convert frontend inference payload to compute API messages format.
+
+    Accepts both:
+      - OpenAI-style: {messages: [{role, content}, ...], temperature, ...}
+      - Frontend-style: {user_question, system_prompt, temperature, ...}
+    """
+    if payload.get("messages"):
+        messages = payload["messages"]
+        # messages already in OpenAI format; pass through with optional system prompt
+        if payload.get("system_prompt") and not any(m.get("role") == "system" for m in messages):
+            messages = [{"role": "system", "content": payload["system_prompt"]}] + list(messages)
+    else:
+        messages = []
+        if payload.get("system_prompt"):
+            messages.append({"role": "system", "content": payload["system_prompt"]})
+        question = payload.get("user_question") or payload.get("question") or ""
+        if question:
+            messages.append({"role": "user", "content": question})
+    return {
+        "messages": messages,
+        "temperature": float(payload.get("temperature", 0.7)),
+        "top_p": float(payload.get("top_p", 0.95)),
+        "max_new_tokens": int(payload.get("max_tokens", 2048)),
+        "do_sample": bool(payload.get("do_sample", True)),
+    }
+
+
+async def _stream_chat_proxy(payload: dict[str, Any]) -> StreamingResponse:
+    """Common SSE streaming proxy: convert payload → forward to compute node → stream back."""
+    store = get_platform_store()
+    node = _select_first_online_node(store)
+    if not node:
+        return StreamingResponse(
+            iter(['data: {"error": "no online compute node available for inference"}\n\n']),
+            media_type="text/event-stream",
+        )
+    client = ComputeNodeClient(node["api_base_url"])
+    compute_payload = _build_messages_payload(payload)
+
+    async def stream_proxy():
+        async with httpx.AsyncClient(timeout=300) as http:
+            url = f"{node['api_base_url'].rstrip('/')}{client.route_prefix}/inference/chat/stream"
+            try:
+                async with http.stream("POST", url, json=compute_payload, headers=client.headers()) as resp:
+                    if resp.status_code >= 400:
+                        yield f'data: {{"error": "compute node returned {resp.status_code}"}}\n\n'.encode()
+                        return
+                    async for chunk in resp.aiter_bytes():
+                        yield chunk
+            except Exception as exc:
+                yield f'data: {{"error": "stream proxy failed: {exc}"}}\n\n'.encode()
+
+    return StreamingResponse(stream_proxy(), media_type="text/event-stream")
+
+
 def fail(status_code: int, message: str) -> HTTPException:
     return HTTPException(status_code=status_code, detail={"code": status_code, "message": message, "data": None})
 
@@ -856,6 +912,7 @@ async def upload_dataset_files(
 ) -> dict[str, Any]:
     created: list[dict[str, Any]] = []
     compute_sync: list[dict[str, Any]] = []
+    pending_sync: list[tuple[str, str, bytes]] = []
     store = get_platform_store()
     try:
         store.dataset(dataset_id)
@@ -867,16 +924,18 @@ async def upload_dataset_files(
             content = raw.decode("utf-8", errors="replace")
             created_file = store.add_dataset_file(conn, dataset_id, file.filename or "upload.jsonl", content)
             created.append(created_file)
-            if sync_to_compute:
-                compute_sync.extend(
-                    await _sync_dataset_file_to_compute_nodes(
-                        store,
-                        dataset_id,
-                        created_file["id"],
-                        created_file["name"],
-                        raw,
-                    )
+            pending_sync.append((created_file["id"], created_file["name"], raw))
+    if sync_to_compute:
+        for file_id, file_name, raw in pending_sync:
+            compute_sync.extend(
+                await _sync_dataset_file_to_compute_nodes(
+                    store,
+                    dataset_id,
+                    file_id,
+                    file_name,
+                    raw,
                 )
+            )
     return ok({"files": created, "compute_sync": compute_sync})
 
 
@@ -1212,7 +1271,23 @@ async def model_eval_list(current_user: dict = Depends(get_current_user)) -> dic
 @router.get("/model-eval/{task_id}")
 async def model_eval_detail(task_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
     try:
-        task = get_platform_store().eval_task(task_id)
+        store = get_platform_store()
+        task = store.eval_task(task_id)
+        if task.get("compute_job_id") and task.get("compute_node_id") and task.get("status") in {"queued", "running", "completed"}:
+            node = next(
+                (n for n in store.compute_nodes() if n["id"] == task.get("compute_node_id")),
+                None,
+            )
+            if node:
+                try:
+                    client = ComputeNodeClient(node["api_base_url"])
+                    job = await client.get_job(task["compute_job_id"])
+                    result_content = None
+                    if job.get("status") == "completed" and not task.get("samples"):
+                        result_content = await fetch_eval_result_content(client, node, job)
+                    task = store.apply_eval_job_result(task_id, job, result_content)
+                except Exception:
+                    pass
     except KeyError:
         raise fail(404, "eval task not found")
     if not has_resource_access("eval", task_id, current_user, "read"):
@@ -1222,8 +1297,149 @@ async def model_eval_detail(task_id: str, current_user: dict = Depends(get_curre
 
 @router.post("/model-eval/start")
 async def model_eval_start(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
-    task = get_platform_store().create_eval_task(payload)
-    return ok({"task_id": task["id"], **task})
+    """Start an evaluation task: submit eval job to compute node."""
+    store = get_platform_store()
+    # 1. Create eval task record
+    task = store.create_eval_task({**payload, "status": "pending"})
+
+    # 2. Resolve model path (supports both regular models and trained models)
+    model_id = str(payload.get("model_id", ""))
+    model_path = ""
+    adapter_path = payload.get("adapter_path", "")
+    try:
+        db_model = store.model(model_id)
+        model_path = db_model.get("path", "")
+    except KeyError:
+        # Try trained_models table (IDs prefixed with tm_)
+        trained = next((m for m in store.trained_models() if m["id"] == model_id), None)
+        if trained:
+            merged_path = trained.get("merged_path", "")
+            base_path = trained.get("base_model_path", "")
+            if trained.get("merged") and merged_path:
+                # Merged model: use merged_path as model, no adapter needed
+                model_path = merged_path
+            elif base_path:
+                # Unmerged: use base model + adapter checkpoint
+                model_path = base_path
+                if merged_path:
+                    adapter_path = merged_path
+            else:
+                model_path = merged_path or base_path
+    if not model_path:
+        store.update_eval_task(task["id"], {"status": "failed", "error": "model not found or no path"})
+        return ok({"task_id": task["id"], "status": "failed", "error": "model not found or no path"})
+
+    # 3. Resolve dataset file
+    dataset_id = str(payload.get("dataset_id", ""))
+    dataset_path = ""
+    try:
+        ds_files = store.training_dataset_files(dataset_id)
+        if ds_files:
+            dataset_path = ds_files[0].get("local_path") or ds_files[0].get("name", "")
+    except Exception:
+        pass
+    if not dataset_path:
+        # Try to get file content and sync to compute
+        try:
+            ds = store.dataset(dataset_id)
+            for f in ds.get("files", []):
+                if f.get("content"):
+                    dataset_path = f.get("name", f"dataset_{dataset_id}.jsonl")
+                    break
+        except KeyError:
+            pass
+    if not dataset_path:
+        store.update_eval_task(task["id"], {"status": "failed", "error": "dataset not found or no files"})
+        return ok({"task_id": task["id"], "status": "failed", "error": "dataset not found or no files"})
+
+    # 4. Resolve dimension config
+    dimension_id = str(payload.get("dimension_id", ""))
+    dimension_cfg: dict[str, Any] = {}
+    if dimension_id:
+        try:
+            dim = store.dimension(dimension_id)
+            # Resolve eval model API config
+            eval_model_name = dim.get("eval_model", "")
+            api_url = ""
+            api_key = ""
+            if eval_model_name:
+                try:
+                    eval_model = store.model(eval_model_name) if eval_model_name.startswith("m_") else store.model_by_name(eval_model_name)
+                    api_url = eval_model.get("api_url", "")
+                    api_key = eval_model.get("api_key", "")
+                except (KeyError, Exception):
+                    pass
+            dimension_cfg = {
+                "type": dim.get("type", ""),
+                "eval_model": eval_model_name,
+                "eval_method": dim.get("eval_method", ""),
+                "eval_prompt": dim.get("eval_prompt", ""),
+                "api_url": api_url,
+                "api_key": api_key,
+                "score_min": dim.get("score_min", 0),
+                "score_max": dim.get("score_max", 5),
+                "pass_threshold": dim.get("pass_threshold", 3),
+            }
+        except KeyError:
+            pass
+
+    # 5. Select compute node
+    node = _select_first_online_node(store)
+    if not node:
+        store.update_eval_task(task["id"], {"status": "failed", "error": "no online compute node"})
+        return ok({"task_id": task["id"], "status": "failed", "error": "no online compute node"})
+
+    # 6. Build eval job payload
+    output_dir = f"/data/yg-ft/outputs/{task['id']}"
+    job_payload = {
+        "id": f"eval_{task['id']}",
+        "name": task.get("eval_task_name", task["id"]),
+        "engine": "eval",
+        "model_name_or_path": model_path,
+        "adapter_name_or_path": adapter_path,
+        "template": payload.get("template", "qwen"),
+        "dataset_path": dataset_path,
+        "output_dir": output_dir,
+        "basic_metrics": payload.get("basic_metrics", {}),
+        "dimension": dimension_cfg,
+        "gpus": [int(payload.get("gpu_id", 0))],
+        "temperature": payload.get("temperature", 0.1),
+        "max_new_tokens": payload.get("max_new_tokens", 512),
+        "compute_node_id": node["id"],
+    }
+
+    # 7. Submit to compute node via create_job (uses engine="eval" path)
+    try:
+        client = ComputeNodeClient(node["api_base_url"])
+        # Sync dataset file to compute node if needed
+        if not dataset_path.startswith("/"):
+            try:
+                ds_files = store.training_dataset_files(dataset_id)
+                if ds_files and ds_files[0].get("content"):
+                    upload_result = await client.upload_file(
+                        ds_files[0].get("name", "eval_data.jsonl"),
+                        ds_files[0]["content"].encode("utf-8"),
+                        f"datasets/{dataset_id}/{ds_files[0].get('name', 'eval_data.jsonl')}",
+                        resource_type="dataset",
+                        resource_id=dataset_id,
+                    )
+                    job_payload["dataset_path"] = upload_result.get("local_path", dataset_path)
+            except Exception:
+                pass
+
+        job = await client.create_job(job_payload)
+        store.update_eval_task(task["id"], {
+            "status": "running",
+            "compute_job_id": job.get("id"),
+            "compute_node_id": node["id"],
+            "output_dir": output_dir,
+        })
+        if job.get("status") in {"queued", "running"}:
+            store.mark_inference_loaded(node["id"])
+        return ok({"task_id": task["id"], "status": "running", "job": job})
+    except Exception as exc:
+        store.update_eval_task(task["id"], {"status": "failed", "error": str(exc)})
+        return ok({"task_id": task["id"], "status": "failed", "error": str(exc)})
 
 
 @router.delete("/model-eval/{task_id}")
@@ -1298,8 +1514,27 @@ async def model_compare_detail(task_id: str) -> dict[str, Any]:
         raise fail(404, "compare task not found")
 
 
+async def _unload_from_compute_node() -> dict[str, Any]:
+    """Best-effort unload the inference model from the first online compute node."""
+    store = get_platform_store()
+    # Clear all inference tracking — only one model can be loaded at a time
+    for node in store.compute_nodes():
+        store.mark_inference_unloaded(node["id"])
+    node = _select_first_online_node(store)
+    if not node:
+        return {"unloaded": False, "error": "no online compute node"}
+    try:
+        client = ComputeNodeClient(node["api_base_url"])
+        result = await client._request("POST", "/inference/unload", json_data={})
+        return result
+    except Exception as exc:
+        return {"unloaded": False, "error": str(exc)}
+
+
 @router.delete("/model-compare/{task_id}")
 async def model_compare_delete(task_id: str) -> dict[str, Any]:
+    # 删除前先释放算力节点上的模型
+    await _unload_from_compute_node()
     get_platform_store().delete_compare_task(task_id)
     return ok({"deleted": task_id})
 
@@ -1329,26 +1564,56 @@ async def model_compare_update_load_status(task_id: str, payload: dict[str, Any]
 
 @router.post("/model-compare/{task_id}/load")
 async def model_compare_load(task_id: str) -> dict[str, Any]:
+    """真正加载模型到算力节点（不再使用假 PID/端口）。"""
     try:
-        task = get_platform_store().compare_task(task_id)
+        store = get_platform_store()
+        task = store.compare_task(task_id)
         models = task.get("models") or []
         if isinstance(models, str):
             try:
                 models = json.loads(models)
             except json.JSONDecodeError:
                 models = []
-        loaded_models = [
-            {
-                "model_id": item.get("model_id"),
-                "model_name": item.get("model_name"),
-                "status": "ready",
-                "pid": 45000 + index,
-                "port": item.get("port") or 18000 + index,
+        # 选取在线算力节点
+        node = _select_first_online_node(store)
+        if not node:
+            return ok({"status": "failed", "error": "no online compute node"})
+        client = ComputeNodeClient(node["api_base_url"])
+        loaded_models = []
+        for item in models:
+            if not isinstance(item, dict):
+                continue
+            model_path = item.get("model_path", "")
+            if not model_path:
+                # 尝试从模型库获取路径
+                model_id = item.get("model_id", "")
+                try:
+                    db_model = store.model(model_id)
+                    model_path = db_model.get("path", "")
+                except KeyError:
+                    pass
+            if not model_path:
+                loaded_models.append({**item, "status": "error", "error": "model_path not found"})
+                continue
+            # 真正调用算力节点加载模型
+            load_payload = {
+                "model_name_or_path": model_path,
+                "template": item.get("template", "qwen"),
             }
-            for index, item in enumerate(models)
-            if isinstance(item, dict)
-        ]
-        return ok(get_platform_store().update_compare_task(task_id, {"status": "loaded", "load_status": {"loaded_models": loaded_models}}))
+            if item.get("adapter_path"):
+                load_payload["adapter_name_or_path"] = item["adapter_path"]
+            try:
+                result = await client._request("POST", "/inference/load", json_data=load_payload)
+                if result.get("loaded"):
+                    store.mark_inference_loaded(node["id"])
+                    loaded_models.append({**item, "status": "ready"})
+                else:
+                    loaded_models.append({**item, "status": "error", "error": result.get("error", "load failed")})
+            except Exception as exc:
+                loaded_models.append({**item, "status": "error", "error": str(exc)})
+        status = "loaded" if any(m.get("status") == "ready" for m in loaded_models) else "failed"
+        updated = store.update_compare_task(task_id, {"status": status, "load_status": {"loaded_models": loaded_models}})
+        return ok(updated)
     except KeyError:
         raise fail(404, "compare task not found")
 
@@ -1356,7 +1621,11 @@ async def model_compare_load(task_id: str) -> dict[str, Any]:
 @router.post("/model-compare/{task_id}/unload")
 async def model_compare_unload(task_id: str) -> dict[str, Any]:
     try:
-        return ok(get_platform_store().update_compare_task(task_id, {"status": "pending", "load_status": {"loaded_models": []}}))
+        store = get_platform_store()
+        # 真正释放算力节点上的模型资源
+        unload_result = await _unload_from_compute_node()
+        updated = store.update_compare_task(task_id, {"status": "pending", "load_status": {"loaded_models": []}})
+        return ok({"task": updated, "unload": unload_result})
     except KeyError:
         raise fail(404, "compare task not found")
 
@@ -1368,18 +1637,23 @@ async def model_compare_start_model(task_id: str, payload: dict[str, Any] = Body
 
 @router.post("/model-compare/chat-with-port")
 async def model_compare_chat_with_port(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
-    question = ""
-    for message in payload.get("messages") or []:
-        if message.get("role") == "user":
-            question = str(message.get("content") or "")
-    content = f"当前后端已收到推理请求：{question[:120]}"
-    return ok({"response": content, "content": content})
+    """Proxy non-streaming chat to the compute node running the inference model."""
+    store = get_platform_store()
+    node = _select_first_online_node(store)
+    if not node:
+        return ok({"response": "no online compute node available for inference", "request": payload})
+    try:
+        client = ComputeNodeClient(node["api_base_url"])
+        result = await client._request("POST", "/inference/chat", json_data=_build_messages_payload(payload))
+        return ok(result)
+    except Exception as exc:
+        return ok({"response": f"inference failed: {exc}", "request": payload})
 
 
 @router.post("/model-compare/stream-chat")
-async def model_compare_stream_chat(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
-    question = payload.get("user_question") or payload.get("question") or ""
-    return ok({"response": f"当前后端已收到流式推理请求：{str(question)[:120]}"})
+async def model_compare_stream_chat(payload: dict[str, Any] = Body(...)) -> StreamingResponse:
+    """Stream chat from the compute node (SSE proxy)."""
+    return await _stream_chat_proxy(payload)
 
 
 @router.post("/model-chat/batch")
@@ -1396,7 +1670,7 @@ async def model_chat_local(payload: dict[str, Any] = Body(...)) -> dict[str, Any
         return ok({"response": "no online compute node available for inference", "request": payload})
     try:
         client = ComputeNodeClient(node["api_base_url"])
-        result = await client._request("POST", "/inference/chat", json_data=payload)
+        result = await client._request("POST", "/inference/chat", json_data=_build_messages_payload(payload))
         return ok(result)
     except Exception as exc:
         return ok({"response": f"inference failed: {exc}", "request": payload})
@@ -1405,28 +1679,15 @@ async def model_chat_local(payload: dict[str, Any] = Body(...)) -> dict[str, Any
 @router.post("/model-chat/local/chat/stream")
 async def model_chat_local_stream(payload: dict[str, Any] = Body(...)) -> StreamingResponse:
     """Stream chat from the compute node."""
-    store = get_platform_store()
-    node = _select_first_online_node(store)
-    if not node:
-        return StreamingResponse(
-            iter(['data: {"error": "no online compute node"}\n\n']),
-            media_type="text/event-stream",
-        )
-    client = ComputeNodeClient(node["api_base_url"])
-
-    async def stream_proxy():
-        async with httpx.AsyncClient(timeout=300) as http:
-            url = f"{node['api_base_url'].rstrip('/')}/modelTF/inference/chat/stream"
-            async with http.stream("POST", url, json=payload, headers=client.headers()) as resp:
-                async for chunk in resp.aiter_bytes():
-                    yield chunk
-
-    return StreamingResponse(stream_proxy(), media_type="text/event-stream")
+    return await _stream_chat_proxy(payload)
 
 
 @router.post("/model-chat/local/preload")
 async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
     """Load a model on the compute node for inference."""
+    model_path = (payload.get("model_name_or_path") or "").strip()
+    if not model_path:
+        return ok({"loaded": False, "error": "model_name_or_path is required"})
     store = get_platform_store()
     node = _select_first_online_node(store)
     if not node:
@@ -1434,6 +1695,8 @@ async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[
     try:
         client = ComputeNodeClient(node["api_base_url"])
         result = await client._request("POST", "/inference/load", json_data=payload)
+        if result.get("loaded"):
+            store.mark_inference_loaded(node["id"])
         return ok(result)
     except Exception as exc:
         return ok({"loaded": False, "error": str(exc)})
@@ -1443,6 +1706,9 @@ async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[
 async def model_chat_local_unload() -> dict[str, Any]:
     """Unload the inference model from the compute node."""
     store = get_platform_store()
+    # Clear all inference tracking
+    for n in store.compute_nodes():
+        store.mark_inference_unloaded(n["id"])
     node = _select_first_online_node(store)
     if not node:
         return ok({"unloaded": False, "error": "no online compute node"})
@@ -1472,6 +1738,9 @@ async def model_chat_local_status() -> dict[str, Any]:
 @router.post("/model-chat/trained/preload")
 async def model_chat_trained_preload(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
     """Load a trained model (base + adapter) on the compute node for inference."""
+    model_path = (payload.get("model_name_or_path") or "").strip()
+    if not model_path:
+        return ok({"loaded": False, "error": "model_name_or_path is required"})
     store = get_platform_store()
     node = _select_first_online_node(store)
     if not node:
@@ -1479,6 +1748,8 @@ async def model_chat_trained_preload(payload: dict[str, Any] = Body(...)) -> dic
     try:
         client = ComputeNodeClient(node["api_base_url"])
         result = await client._request("POST", "/inference/load", json_data=payload)
+        if result.get("loaded"):
+            store.mark_inference_loaded(node["id"])
         return ok(result)
     except Exception as exc:
         return ok({"loaded": False, "error": str(exc)})
@@ -1517,6 +1788,16 @@ async def update_compute_node(node_id: str, payload: dict[str, Any] = Body(...))
         raise fail(400, str(exc))
 
 
+@router.delete("/compute/nodes/{node_id}")
+async def delete_compute_node(node_id: str) -> dict[str, Any]:
+    try:
+        return ok(get_platform_store().delete_compute_node(node_id))
+    except KeyError:
+        raise fail(404, "compute node not found")
+    except ValueError as exc:
+        raise fail(400, str(exc))
+
+
 @router.post("/compute/nodes/{node_id}/test-connection")
 async def test_compute_node(node_id: str) -> dict[str, Any]:
     store = get_platform_store()

backend/app/db/platform_store.py

@@ -88,6 +88,22 @@ def parse_size_bytes(value: Any) -> int:
     return max(0, round(amount * _SIZE_UNIT_BYTES[unit]))
 
 
+def count_dataset_records(content: str) -> int:
+    text = (content or "").strip()
+    if not text:
+        return 0
+
+    try:
+        value = json.loads(text)
+        if isinstance(value, list):
+            return len(value)
+        return 1
+    except (TypeError, ValueError, json.JSONDecodeError):
+        pass
+
+    return len([line for line in text.splitlines() if line.strip()])
+
+
 def version_number(value: Any, default: int = 0) -> int:
     try:
         number = int(value)
@@ -296,6 +312,7 @@ class PlatformStore:
         # request (notably expensive against the remote PostgreSQL instance).
         # TCP keepalive 让操作系统持续保活连接，抵抗远程库空闲静默断连。
         pool_kwargs = {
+            "connect_timeout": 5,
             "keepalives": 1,
             "keepalives_idle": 10,
             "keepalives_interval": 5,
@@ -320,6 +337,19 @@ class PlatformStore:
         self._pool.open()
         self.ensure_schema()
         self.ensure_seed_data()
+        # Track which compute nodes have an active inference model loaded
+        self._inference_nodes: set[str] = set()
+
+    # ── inference node tracking ────────────────────────────────────
+
+    def mark_inference_loaded(self, node_id: str) -> None:
+        self._inference_nodes.add(node_id)
+
+    def mark_inference_unloaded(self, node_id: str) -> None:
+        self._inference_nodes.discard(node_id)
+
+    def is_inference_loaded(self, node_id: str) -> bool:
+        return node_id in self._inference_nodes
 
     @contextmanager
     def connect(self) -> Iterator["PgConnection"]:
@@ -1298,6 +1328,7 @@ class PlatformStore:
             file_size_bytes = int(file_row.get("size_bytes") or 0)
             if file_size_bytes <= 0:
                 file_size_bytes = parse_size_bytes(file_row.get("size"))
+            file_record_count = int(file_row.get("record_count") or 0)
             decoded_files.append(
                 {
                     "id": file_row["id"],
@@ -1306,7 +1337,7 @@ class PlatformStore:
                     "size_bytes": file_size_bytes,
                     **dataset_file_version_summary(file_row),
                     "create_time": file_row["create_time"],
-                    "record_count": int(file_row.get("record_count") or 0),
+                    "record_count": file_record_count,
                     "split": metadata.get("file_split"),
                 }
             )
@@ -1324,6 +1355,9 @@ class PlatformStore:
             total_size_bytes = int(row.get("size_bytes") or 0)
             if total_size_bytes <= 0:
                 total_size_bytes = parse_size_bytes(row.get("size"))
+        total_record_count = sum(int(item.get("record_count") or 0) for item in decoded_files)
+        if not decoded_files:
+            total_record_count = int(row.get("record_count") or row.get("count") or 0)
         current_version_nos = sorted(
             {
                 int(item["current_version_no"])
@@ -1333,6 +1367,8 @@ class PlatformStore:
         )
         return {
             **dict(row),
+            "count": total_record_count,
+            "record_count": total_record_count,
             "size_bytes": total_size_bytes,
             "current_version_no": (
                 current_version_nos[0] if len(current_version_nos) == 1 else None
@@ -1407,7 +1443,7 @@ class PlatformStore:
         version_id = f"{file_id}_v1"
         size_bytes = len(content.encode("utf-8"))
         size = f"{size_bytes} B"
-        record_count = len([line for line in content.splitlines() if line.strip()])
+        record_count = count_dataset_records(content)
         version = {
             "id": version_id,
             "version": 1,
@@ -1440,10 +1476,18 @@ class PlatformStore:
         )
         conn.execute(
             """UPDATE datasets
-               SET count=count+?, record_count=record_count+?,
-                   size_bytes=size_bytes+?, size=((size_bytes+?)::text || ' B')
+               SET count=stats.record_count,
+                   record_count=stats.record_count,
+                   size_bytes=stats.size_bytes,
+                   size=(stats.size_bytes::text || ' B')
+               FROM (
+                   SELECT COALESCE(SUM(record_count), 0) AS record_count,
+                          COALESCE(SUM(size_bytes), 0) AS size_bytes
+                   FROM dataset_files
+                   WHERE dataset_id=?
+               ) stats
                WHERE id=?""",
-            (record_count, record_count, size_bytes, size_bytes, dataset_id),
+            (dataset_id, dataset_id),
         )
         return {
             "id": file_id,
@@ -1548,19 +1592,57 @@ class PlatformStore:
             row = conn.execute("SELECT * FROM dataset_files WHERE id=?", (file_id,)).fetchone()
             if not row:
                 raise KeyError(file_id)
+            content = payload.get("content", "")
+            size_bytes = len(content.encode("utf-8"))
+            record_count = count_dataset_records(content)
             versions = json_loads(row["versions"], [])
             version = {
                 "id": f"{file_id}_v{len(versions) + 1}",
                 "version": len(versions) + 1,
+                "version_no": len(versions) + 1,
                 "create_time": utcnow(),
                 "description": payload.get("description", "online edit"),
+                "size_bytes": size_bytes,
+                "record_count": record_count,
             }
             versions.append(version)
             conn.execute(
-                "UPDATE dataset_files SET content=?, active_version_id=?, versions=? WHERE id=?",
-                (payload.get("content", ""), version["id"], json_dumps(versions), file_id),
+                """
+                UPDATE dataset_files
+                SET content=?, active_version_id=?, current_version_id=?, versions=?,
+                    size_bytes=?, size=?, record_count=?, version_no=?
+                WHERE id=?
+                """,
+                (
+                    content,
+                    version["id"],
+                    version["id"],
+                    json_dumps(versions),
+                    size_bytes,
+                    f"{size_bytes} B",
+                    record_count,
+                    version["version_no"],
+                    file_id,
+                ),
             )
-            return {"version": version, "content": payload.get("content", "")}
+            conn.execute(
+                """UPDATE datasets
+                   SET count=stats.record_count,
+                       record_count=stats.record_count,
+                       size_bytes=stats.size_bytes,
+                       size=(stats.size_bytes::text || ' B')
+                   FROM (
+                       SELECT dataset_id,
+                              COALESCE(SUM(record_count), 0) AS record_count,
+                              COALESCE(SUM(size_bytes), 0) AS size_bytes
+                       FROM dataset_files
+                       WHERE dataset_id=(SELECT dataset_id FROM dataset_files WHERE id=?)
+                       GROUP BY dataset_id
+                   ) stats
+                   WHERE datasets.id=stats.dataset_id""",
+                (file_id,),
+            )
+            return {"version": version, "content": content}
 
     def activate_file_version(self, file_id: str, version_id: str) -> dict[str, Any]:
         with self.connect() as conn:
@@ -2114,10 +2196,63 @@ class PlatformStore:
             )
         return self.eval_task(task_id)
 
+    def update_eval_task(self, task_id: str, updates: dict[str, Any]) -> dict[str, Any]:
+        """Update fields in an eval task's payload without replacing the whole record."""
+        task = self.eval_task(task_id)
+        merged = {**task, **updates}
+        with self.connect() as conn:
+            conn.execute(
+                "UPDATE eval_tasks SET payload=?, status=? WHERE id=?",
+                (json_dumps(merged), merged.get("status", task.get("status", "pending")), task_id),
+            )
+        return self.eval_task(task_id)
+
     def delete_eval_task(self, task_id: str) -> None:
         with self.connect() as conn:
             conn.execute("DELETE FROM eval_tasks WHERE id=?", (task_id,))
 
+    def running_eval_tasks(self) -> list[dict[str, Any]]:
+        """Return eval tasks that have been submitted to a compute node and are still running."""
+        return [
+            task for task in self.eval_tasks()
+            if task.get("compute_job_id") and task.get("status") in {"queued", "running"}
+        ]
+
+    def apply_eval_job_result(self, task_id: str, job: dict[str, Any], result_content: dict[str, Any] | None = None) -> dict[str, Any]:
+        """Sync a compute job status/result back to an eval task."""
+        task = self.eval_task(task_id)
+        job_status = str(job.get("status", ""))
+        status_map = {"queued": "running", "running": "running", "completed": "completed",
+                       "failed": "failed", "stopped": "stopped"}
+        new_status = status_map.get(job_status, job_status or task.get("status", "pending"))
+        updates: dict[str, Any] = {
+            "status": new_status,
+            "progress": int(job.get("progress", 0)),
+            "output_dir": job.get("output_dir", task.get("output_dir", "")),
+        }
+        # On completion, populate results from eval_results.json content
+        if new_status == "completed" and result_content:
+            updates.update({
+                "overall_score": result_content.get("overall_score", 0),
+                "overall_score_max": result_content.get("overall_score_max", 100),
+                "overall_evaluation": result_content.get("overall_evaluation", ""),
+                "improvement_suggestions": result_content.get("improvement_suggestions", []),
+                "dimension_summary": result_content.get("dimension_summary", []),
+                "samples": result_content.get("samples", []),
+                "sample_count": result_content.get("sample_count", 0),
+                "completed_count": result_content.get("completed_count", 0),
+                "passed_count": result_content.get("passed_count", 0),
+                "basic_metrics": result_content.get("basic_metrics", {}),
+                "score": result_content.get("overall_score", 0),
+                "completed_time": utcnow(),
+            })
+        elif new_status in {"failed", "stopped"}:
+            updates.update({
+                "error": job.get("error") or task.get("error") or "",
+                "completed_time": utcnow(),
+            })
+        return self.update_eval_task(task_id, updates)
+
     def dimensions(self) -> list[dict[str, Any]]:
         with self.connect() as conn:
             rows = conn.execute("SELECT * FROM eval_dimensions ORDER BY create_time DESC").fetchall()
@@ -2284,6 +2419,15 @@ class PlatformStore:
         ).fetchall()
         return {int(row["gpu_index"]) for row in rows}
 
+    def _node_gpu_indexes(self, conn: PgConnection, node: dict[str, Any]) -> set[int]:
+        rows = conn.execute("SELECT gpu_index FROM gpus WHERE node_id=?", (node["id"],)).fetchall()
+        if rows:
+            return {int(row["gpu_index"]) for row in rows}
+        return set(range(max(0, int(node.get("gpu_count") or 0))))
+
+    def _node_capacity(self, node: dict[str, Any]) -> int:
+        return max(1, int(node.get("max_parallel_jobs") or 1), int(node.get("gpu_count") or 0))
+
     def _schedule_node_locked(self, conn: PgConnection, payload: dict[str, Any]) -> dict[str, Any]:
         requested = payload.get("requested_node_id") or payload.get("compute_node_id")
         requested_gpus = [int(item) for item in payload.get("gpus") or []]
@@ -2291,13 +2435,15 @@ class PlatformStore:
         candidates = [
             n
             for n in nodes
-            if n["enabled"] and n["scheduler_status"] == "online" and n["current_running_jobs"] < n["max_parallel_jobs"]
+            if n["enabled"] and n["scheduler_status"] == "online" and n["current_running_jobs"] < self._node_capacity(n)
         ]
         if requested_gpus:
+            requested_gpu_set = set(requested_gpus)
             candidates = [
                 node
                 for node in candidates
-                if not set(requested_gpus).intersection(self._active_gpu_indexes(conn, node["id"]))
+                if requested_gpu_set.issubset(self._node_gpu_indexes(conn, node))
+                and not requested_gpu_set.intersection(self._active_gpu_indexes(conn, node["id"]))
             ]
         if requested:
             selected = next((n for n in candidates if n["id"] == requested), None)
@@ -2312,8 +2458,8 @@ class PlatformStore:
                     reason = "disabled"
                 elif node["scheduler_status"] != "online":
                     reason = f"status={node['scheduler_status']}"
-                elif node["current_running_jobs"] >= node["max_parallel_jobs"]:
-                    reason = f"capacity full {node['current_running_jobs']}/{node['max_parallel_jobs']}"
+                elif node["current_running_jobs"] >= self._node_capacity(node):
+                    reason = f"capacity full {node['current_running_jobs']}/{self._node_capacity(node)}"
                 else:
                     reason = "not selected"
                 reasons.append(f"{node['code']}({reason})")
@@ -2664,6 +2810,24 @@ class PlatformStore:
             )
         return next(node for node in self.compute_nodes() if node["id"] == node_id)
 
+    def delete_compute_node(self, node_id: str) -> dict[str, Any]:
+        with self.connect() as conn:
+            node = conn.execute("SELECT * FROM compute_nodes WHERE id=?", (node_id,)).fetchone()
+            if not node:
+                raise KeyError(node_id)
+            active = conn.execute(
+                """
+                SELECT COUNT(*) AS cnt
+                FROM fine_tune_tasks
+                WHERE compute_node_id=? AND status IN ('syncing','queued','running')
+                """,
+                (node_id,),
+            ).fetchone()
+            if active and int(active["cnt"] or 0) > 0:
+                raise ValueError("compute node has active training tasks")
+            conn.execute("DELETE FROM compute_nodes WHERE id=?", (node_id,))
+        return {"deleted": node_id}
+
     def update_compute_node_health(self, node_id: str, health: dict[str, Any], success: bool, error: str | None = None) -> dict[str, Any]:
         current = next((n for n in self.compute_nodes() if n["id"] == node_id), None)
         if not current:
@@ -2749,6 +2913,11 @@ class PlatformStore:
                 )
                 busy = task is not None and task.get("status") == "running"
                 reserved = task is not None and task.get("status") in {"syncing", "queued"}
+                # Also mark GPU as busy if an inference model is loaded on this node
+                inference_busy = self.is_inference_loaded(row["node_id"])
+                if inference_busy and not busy:
+                    busy = True
+                    reserved = False
                 memory_used = round(row["memory_total_gb"] * (0.72 if busy else 0.18 if reserved else 0.04), 1)
                 gpu_percent = 86 if busy else 22 if reserved else 3
                 memory_total = float(row["memory_total_gb"] or 0)
@@ -2827,11 +2996,12 @@ class PlatformStore:
         }
 
     def health_metrics(self) -> dict[str, float]:
-        info = self.system_info()
+        # Health checks must stay lightweight. The Docker healthcheck and page
+        # refresh probes should not wait on dashboard/GPU/database aggregation.
         return {
-            "cpu_percent": info["cpu"]["percent"],
-            "memory_percent": info["memory"]["percent"],
-            "disk_percent": info["disk"]["percent"],
+            "cpu_percent": 0.0,
+            "memory_percent": 0.0,
+            "disk_percent": 0.0,
         }
 
     def queue(self) -> list[dict[str, Any]]:

backend/app/modules/compute_gateway/client.py

@@ -207,7 +207,8 @@ class ComputeNodeClient:
             "resource_id": resource_id or "",
         }
         files = {"file": (filename, content)}
-        async with httpx.AsyncClient(timeout=max(self.timeout, 60), headers=self.headers()) as client:
+        timeout = httpx.Timeout(max(self.timeout, 60), connect=self.timeout)
+        async with httpx.AsyncClient(timeout=timeout, headers=self.headers()) as client:
             response = await client.post(
                 _join_url(self.api_base_url, f"{self.route_prefix}/compute/files/upload"),
                 data=data,

backend/app/modules/compute_gateway/sync.py

@@ -10,6 +10,24 @@ def _node_for_task(task: dict[str, Any]) -> dict[str, Any] | None:
     return next((node for node in get_platform_store().compute_nodes() if node["id"] == task.get("compute_node_id")), None)
 
 
+async def fetch_eval_result_content(client: ComputeNodeClient, node: dict[str, Any], job: dict[str, Any]) -> dict[str, Any] | None:
+    output_dir = job.get("output_dir")
+    if not output_dir:
+        return None
+    full_path = f"{str(output_dir).rstrip('/')}/eval_results.json"
+    data_root = "/data/yg-ft/"
+    if full_path.startswith(data_root):
+        full_path = full_path[len(data_root):]
+    rel_path = full_path.lstrip("/")
+    import httpx
+    url = f"{node['api_base_url'].rstrip('/')}/modelTF/compute/files/read"
+    async with httpx.AsyncClient(timeout=30, headers=client.headers()) as http:
+        response = await http.get(url, params={"path": rel_path})
+        response.raise_for_status()
+        payload = response.json()
+        return payload if isinstance(payload, dict) else None
+
+
 async def poll_compute_jobs_once() -> dict[str, Any]:
     store = get_platform_store()
     synced: list[dict[str, Any]] = []
@@ -48,4 +66,34 @@ async def poll_compute_jobs_once() -> dict[str, Any]:
             standalone_synced.append(store.sync_model_merge_job(record["id"], job))
         except Exception as exc:  # noqa: BLE001 - keep polling other jobs
             failed.append({"job_id": record["id"], "error": str(exc)})
-    return {"synced": len(synced) + len(standalone_synced), "failed": failed, "items": synced, "standalone": standalone_synced}
+
+    # ── Eval job sync ────────────────────────────────────────────────
+    eval_synced = 0
+    for eval_task in store.running_eval_tasks():
+        node = next(
+            (item for item in store.compute_nodes() if item["id"] == eval_task.get("compute_node_id")),
+            None,
+        )
+        if not node:
+            failed.append({"eval_task_id": eval_task["id"], "error": "compute node not found"})
+            continue
+        try:
+            client = ComputeNodeClient(node["api_base_url"])
+            job = await client.get_job(eval_task["compute_job_id"])
+            result_content = None
+            # Try to read eval_results.json from the job output directory
+            if job.get("status") == "completed" and job.get("output_dir"):
+                try:
+                    result_content = await fetch_eval_result_content(client, node, job)
+                except Exception:
+                    pass
+            store.apply_eval_job_result(eval_task["id"], job, result_content)
+            # If job completed, un-mark inference loaded
+            if job.get("status") in {"completed", "failed", "stopped"}:
+                store.mark_inference_unloaded(node["id"])
+            eval_synced += 1
+        except Exception as exc:  # noqa: BLE001
+            failed.append({"eval_task_id": eval_task["id"], "error": str(exc)})
+
+    return {"synced": len(synced) + len(standalone_synced) + eval_synced, "failed": failed,
+            "items": synced, "standalone": standalone_synced, "eval_synced": eval_synced}