YG-Soft/YG_FT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增 compute_gateway、compute_poller、agent 模块，重构前端 dist

Browse Source 
- 新增 backend/app/modules/compute_gateway（client/sync）计算网关模块
- 新增 backend/app/workers/compute_poller 计算轮询 worker
- 新增 compute/agent/process_manager 进程管理器
- 新增 scripts/ 脚本目录
- 更新 Docker 部署配置（app/compute/nginx）
- 更新后端平台 API、数据库 SQL、core 配置
- 更新前端多个视图组件及 API 模块
- 重构 frontend/dist 构建产物（新 hash）
- 更新多项文档

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
wuyongtao
2026-07-22 17:32:59 +08:00
parent 1e438164c1
commit 836343b29e
180 changed files with 3352 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

658
backend/app/api/v1/endpoints/platform.py
View File

@@ -1,12 +1,17 @@
from __future__ import annotations
from typing import Any
import uuid
import json
from pathlib import Path
from typing import Any
from fastapi import APIRouter, Body, File, HTTPException, Query, UploadFile
from fastapi.responses import PlainTextResponse
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
router = APIRouter()
@@ -19,6 +24,77 @@ def fail(status_code: int, message: str) -> HTTPException:
return HTTPException(status_code=status_code, detail={"code": status_code, "message": message, "data": None})
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)
def _task_for_compute_job(job_id: str) -> dict[str, Any] | None:
return next((task for task in get_platform_store().tasks() if task.get("compute_job_id") == job_id), None)
async def _submit_fine_tune_task(store: Any, payload: dict[str, Any]) -> dict[str, Any]:
task_id = str(payload.get("task_id") or payload.get("id") or "")
if task_id and get_settings().compute_mode != "simulator":
try:
preflight = await _fine_tune_preflight(store, task_id, payload, validate=True)
except Exception as exc: # noqa: BLE001 - task has not entered running state yet
raise RuntimeError(f"preflight failed: {exc}") from exc
if not preflight["valid"]:
errors = "; ".join(preflight.get("errors") or ["preflight failed"])
raise RuntimeError(f"preflight failed: {errors}")
task = store.start_task(payload)
if get_settings().compute_mode == "simulator":
return task
node, job_payload = store.build_compute_job_payload(task["id"])
job = await ComputeNodeClient(node["api_base_url"]).create_job(job_payload)
return store.apply_compute_job(task["id"], job)
async def _fine_tune_preflight(
store: Any,
task_id: str,
payload: dict[str, Any] | None = None,
validate: bool = True,
) -> dict[str, Any]:
node, job_payload = store.prepare_compute_job_payload(task_id, payload or {})
if get_settings().compute_mode == "simulator":
preview = {
"valid": True,
"errors": [],
"warnings": ["compute_mode=simulator skips remote compute validation"],
"engine": job_payload.get("engine") or job_payload.get("training_engine") or "llama_factory",
"command": [],
"command_text": "",
"work_dir": "",
"env": {},
"path_checks": [],
}
else:
client = ComputeNodeClient(node["api_base_url"])
preview = await (client.validate_job(job_payload) if validate else client.preview_job(job_payload))
errors = list(preview.get("errors") or [])
warnings = list(preview.get("warnings") or [])
if not node.get("enabled"):
errors.append(f"compute node disabled: {node.get('code')}")
if node.get("scheduler_status") not in {"online", "draining"}:
errors.append(f"compute node not schedulable: {node.get('code')} status={node.get('scheduler_status')}")
return {
"valid": bool(preview.get("valid", not errors)) and not errors,
"errors": errors,
"warnings": warnings,
"node": {
"id": node.get("id"),
"code": node.get("code"),
"name": node.get("name"),
"api_base_url": node.get("api_base_url"),
"scheduler_status": node.get("scheduler_status"),
"gpu_count": node.get("gpu_count"),
},
"job_payload": job_payload,
"preview": preview,
}
@router.post("/login")
async def login(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
user = get_platform_store().login(payload.get("username", ""), payload.get("password", ""))
@@ -84,7 +160,29 @@ async def delete_user(user_id: str, current_username: str | None = Query(default
@router.get("/model-manage/local-models")
async def local_models() -> dict[str, Any]:
models = [{"path": item.get("path") or "", "name": item["name"]} for item in get_platform_store().models()]
store = get_platform_store()
models = [{"path": item.get("path") or "", "name": item["name"], "source": "registered"} for item in store.models()]
seen = {item["path"] for item in models if item.get("path")}
if get_settings().compute_mode != "simulator":
for node in store.compute_nodes():
if not node.get("enabled"):
continue
try:
result = await ComputeNodeClient(node["api_base_url"]).list_files(root="models", directories_only=True)
except Exception:
continue
for item in result.get("items") or []:
path = str(item.get("path") or "")
if not path or path in seen:
continue
seen.add(path)
models.append(
{
"path": path,
"name": item.get("name") or path.rsplit("/", 1)[-1],
"source": f"compute:{node.get('code')}",
}
)
return ok({"models": models})
@@ -95,6 +193,7 @@ async def trained_models() -> dict[str, Any]:
@router.delete("/model-manage/trained-models/{model_id}")
async def delete_trained_model(model_id: str, type: str = Query(default="merged")) -> dict[str, Any]:
get_platform_store().delete_trained_model(model_id)
return ok({"deleted": model_id, "type": type})
@@ -113,7 +212,14 @@ async def model_list() -> dict[str, Any]:
@router.post("/model-manage")
async def create_model(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
return ok(get_platform_store().create_model(payload))
try:
return ok(get_platform_store().create_model(payload))
except KeyError as exc:
raise fail(400, f"missing field: {exc}")
except ValueError as exc:
raise fail(400, str(exc))
except Exception as exc: # noqa: BLE001 - keep API errors visible to deployment smoke checks
raise fail(500, f"create model failed: {exc}")
@router.get("/model-manage/{model_id}")
@@ -199,12 +305,73 @@ async def activate_dataset_version(file_id: str, payload: dict[str, Any] = Body(
@router.delete("/dataset-manage/versions/{file_id}/{version_id}")
async def delete_dataset_version(file_id: str, version_id: str) -> dict[str, Any]:
return ok(get_platform_store().file_versions(file_id))
try:
return ok(get_platform_store().delete_file_version(file_id, version_id))
except KeyError:
raise fail(404, "dataset version not found")
except ValueError as exc:
raise fail(400, str(exc))
async def _sync_dataset_file_to_compute_nodes(
store: Any,
dataset_id: str,
file_id: str,
filename: str,
content: bytes,
) -> list[dict[str, Any]]:
results: list[dict[str, Any]] = []
if get_settings().compute_mode == "simulator":
return results
target_name = Path(filename or f"{file_id}.jsonl").name
target_relative_path = f"datasets/{dataset_id}/{target_name}"
for node in store.compute_nodes():
if not node.get("enabled"):
continue
try:
result = await ComputeNodeClient(node["api_base_url"]).upload_file(
target_name,
content,
target_relative_path,
resource_type="dataset",
resource_id=dataset_id,
)
store.upsert_resource_replica(
node["id"],
"dataset",
dataset_id,
str(result.get("local_path") or ""),
)
results.append(
{
"node_id": node["id"],
"node_code": node.get("code"),
"success": True,
"local_path": result.get("local_path"),
"byte_size": result.get("byte_size"),
"checksum_sha256": result.get("checksum_sha256"),
}
)
except Exception as exc: # noqa: BLE001 - keep upload usable while exposing sync failures
results.append(
{
"node_id": node["id"],
"node_code": node.get("code"),
"success": False,
"error": str(exc),
}
)
return results
@router.post("/dataset-manage/upload/{dataset_id}")
async def upload_dataset_files(dataset_id: str, files: list[UploadFile] = File(default=[])) -> dict[str, Any]:
async def upload_dataset_files(
dataset_id: str,
files: list[UploadFile] = File(default=[]),
sync_to_compute: bool = Query(default=True),
) -> dict[str, Any]:
created: list[dict[str, Any]] = []
compute_sync: list[dict[str, Any]] = []
store = get_platform_store()
try:
store.dataset(dataset_id)
@@ -214,8 +381,19 @@ async def upload_dataset_files(dataset_id: str, files: list[UploadFile] = File(d
for file in files:
raw = await file.read()
content = raw.decode("utf-8", errors="replace")
created.append(store.add_dataset_file(conn, dataset_id, file.filename or "upload.jsonl", content))
return ok({"files": created})
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,
)
)
return ok({"files": created, "compute_sync": compute_sync})
@router.get("/dataset-manage/download/{dataset_id}")
@@ -299,12 +477,45 @@ async def create_fine_tune(payload: dict[str, Any] = Body(...)) -> dict[str, Any
@router.post("/fine-tune/start")
async def start_fine_tune(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
store = get_platform_store()
try:
return ok(get_platform_store().start_task(payload))
return ok(await _submit_fine_tune_task(store, payload))
except KeyError:
raise fail(404, "fine tune task not found")
except RuntimeError as exc:
task_id = str(payload.get("task_id") or payload.get("id") or "")
if task_id:
store.mark_task_failed(task_id, str(exc))
raise fail(409, str(exc))
except Exception as exc: # noqa: BLE001 - mark task failed when remote submit fails
task_id = str(payload.get("task_id") or payload.get("id") or "")
if task_id:
store.mark_task_failed(task_id, str(exc))
raise fail(502, f"submit compute job failed: {exc}")
@router.post("/fine-tune/{task_id}/preflight")
async def fine_tune_preflight(task_id: str, payload: dict[str, Any] | None = Body(default=None)) -> dict[str, Any]:
try:
return ok(await _fine_tune_preflight(get_platform_store(), task_id, payload or {}, validate=True))
except KeyError:
raise fail(404, "fine tune task not found")
except RuntimeError as exc:
raise fail(409, str(exc))
except Exception as exc: # noqa: BLE001 - expose compute validation errors to training create page
raise fail(502, f"compute preflight failed: {exc}")
@router.post("/fine-tune/{task_id}/command-preview")
async def fine_tune_command_preview(task_id: str, payload: dict[str, Any] | None = Body(default=None)) -> dict[str, Any]:
try:
return ok(await _fine_tune_preflight(get_platform_store(), task_id, payload or {}, validate=False))
except KeyError:
raise fail(404, "fine tune task not found")
except RuntimeError as exc:
raise fail(409, str(exc))
except Exception as exc: # noqa: BLE001
raise fail(502, f"compute command preview failed: {exc}")
@router.get("/fine-tune/{task_id}")
@@ -315,6 +526,37 @@ async def fine_tune_detail(task_id: str) -> dict[str, Any]:
raise fail(404, "fine tune task not found")
@router.get("/fine-tune/{task_id}/logs")
async def fine_tune_logs(
task_id: str,
tail_lines: int | None = Query(default=500, ge=1, le=5000),
offset: int | None = Query(default=None, ge=0),
limit: int | None = Query(default=None, ge=1, le=5000),
) -> dict[str, Any]:
store = get_platform_store()
try:
task = store.task(task_id)
except KeyError:
raise fail(404, "fine tune task not found")
if task.get("compute_job_id"):
node = _node_for_task(task)
if node:
try:
logs = await ComputeNodeClient(node["api_base_url"]).job_logs(task["compute_job_id"], tail_lines, offset, limit)
if task.get("status") in {"queued", "running", "failed", "stopped", "completed"}:
try:
job = await ComputeNodeClient(node["api_base_url"]).get_job(task["compute_job_id"])
store.apply_compute_job(task_id, job)
except Exception:
pass
return ok({"source": "compute", **logs})
except Exception as exc: # noqa: BLE001 - keep failure reason visible even when log fetch fails
content = task.get("failure_reason") or f"fetch compute log failed: {exc}"
return ok({"job_id": task.get("compute_job_id"), "source": "task", "file": task.get("log_file") or "", "content": content, "size": f"{len(content.encode('utf-8'))} B"})
content = task.get("failure_reason") or ""
return ok({"job_id": task.get("compute_job_id") or "", "source": "task", "file": task.get("log_file") or "", "content": content, "size": f"{len(content.encode('utf-8'))} B"})
@router.put("/fine-tune/{task_id}")
async def update_fine_tune(task_id: str, payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
try:
@@ -325,8 +567,14 @@ async def update_fine_tune(task_id: str, payload: dict[str, Any] = Body(...)) ->
@router.post("/fine-tune/stop/{task_id}")
async def stop_fine_tune(task_id: str) -> dict[str, Any]:
store = get_platform_store()
try:
return ok(get_platform_store().stop_task(task_id))
task = store.task(task_id)
node = _node_for_task(task)
if task.get("compute_job_id") and node and get_settings().compute_mode != "simulator":
job = await ComputeNodeClient(node["api_base_url"]).stop_job(task["compute_job_id"])
return ok(store.apply_compute_job(task_id, job))
return ok(store.stop_task(task_id))
except KeyError:
raise fail(404, "fine tune task not found")
@@ -336,6 +584,27 @@ async def stop_fine_tune_alt(task_id: str) -> dict[str, Any]:
return await stop_fine_tune(task_id)
@router.post("/fine-tune/{task_id}/retry")
async def retry_fine_tune(task_id: str, payload: dict[str, Any] | None = Body(default=None)) -> dict[str, Any]:
store = get_platform_store()
payload = payload or {}
try:
task = store.task(task_id)
except KeyError:
raise fail(404, "fine tune task not found")
if task["status"] not in {"failed", "stopped"} and not payload.get("force"):
raise fail(409, "only failed or stopped tasks can be retried without force=true")
retry_payload = {**task, **payload, "task_id": task_id, "id": task_id}
store.reset_task_for_retry(task_id, retry_payload)
try:
return ok(await _submit_fine_tune_task(store, retry_payload))
except RuntimeError as exc:
raise fail(409, str(exc))
except Exception as exc: # noqa: BLE001 - mark retry failed when remote submit fails
store.mark_task_failed(task_id, str(exc))
raise fail(502, f"retry fine tune task failed: {exc}")
@router.delete("/fine-tune/{task_id}")
async def delete_fine_tune(task_id: str) -> dict[str, Any]:
get_platform_store().delete_task(task_id)
@@ -358,17 +627,217 @@ async def fine_tune_checkpoints(task_id: str) -> dict[str, Any]:
return ok(checkpoints)
@router.get("/model-eval")
async def model_eval_list() -> dict[str, Any]:
return ok(get_platform_store().eval_tasks())
@router.get("/model-eval/{task_id}")
async def model_eval_detail(task_id: str) -> dict[str, Any]:
try:
return ok(get_platform_store().eval_task(task_id))
except KeyError:
raise fail(404, "eval task not found")
@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})
@router.delete("/model-eval/{task_id}")
async def model_eval_delete(task_id: str) -> dict[str, Any]:
get_platform_store().delete_eval_task(task_id)
return ok({"deleted": task_id})
@router.get("/dimension")
async def dimension_list() -> dict[str, Any]:
return ok(get_platform_store().dimensions())
@router.post("/dimension")
async def dimension_create(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
return ok(get_platform_store().create_dimension(payload))
@router.get("/dimension/{dimension_id}")
async def dimension_detail(dimension_id: str) -> dict[str, Any]:
try:
return ok(get_platform_store().dimension(dimension_id))
except KeyError:
raise fail(404, "dimension not found")
@router.put("/dimension/{dimension_id}")
async def dimension_update(dimension_id: str, payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
try:
return ok(get_platform_store().update_dimension(dimension_id, payload))
except KeyError:
raise fail(404, "dimension not found")
@router.delete("/dimension/{dimension_id}")
async def dimension_delete(dimension_id: str) -> dict[str, Any]:
get_platform_store().delete_dimension(dimension_id)
return ok({"deleted": dimension_id})
@router.get("/model-compare")
async def model_compare_list() -> dict[str, Any]:
return ok(get_platform_store().compare_tasks())
@router.post("/model-compare")
async def model_compare_create(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
task = get_platform_store().create_compare_task(payload)
return ok({"id": task["id"]})
@router.post("/model-compare/all/stop-all")
async def model_compare_stop_all() -> dict[str, Any]:
return ok({"stopped": True})
@router.post("/model-compare/stop-by-pid")
async def model_compare_stop_by_pid(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
return ok({"stopped": True, "pid": payload.get("pid")})
@router.get("/model-compare/{task_id}")
async def model_compare_detail(task_id: str) -> dict[str, Any]:
try:
return ok(get_platform_store().compare_task(task_id))
except KeyError:
raise fail(404, "compare task not found")
@router.delete("/model-compare/{task_id}")
async def model_compare_delete(task_id: str) -> dict[str, Any]:
get_platform_store().delete_compare_task(task_id)
return ok({"deleted": task_id})
@router.get("/model-compare/{task_id}/load-status")
async def model_compare_load_status(task_id: str) -> dict[str, Any]:
try:
task = get_platform_store().compare_task(task_id)
except KeyError:
raise fail(404, "compare task not found")
load_status = task.get("load_status") or {"loaded_models": []}
if isinstance(load_status, str):
try:
load_status = json.loads(load_status)
except json.JSONDecodeError:
load_status = {"loaded_models": []}
return ok({"all_ready": all(item.get("status") in {"ready", "running"} for item in load_status.get("loaded_models", [])), **load_status})
@router.post("/model-compare/{task_id}/load-status")
async def model_compare_update_load_status(task_id: str, payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
try:
return ok(get_platform_store().update_compare_task(task_id, {"load_status": payload.get("load_status") or {"loaded_models": []}}))
except KeyError:
raise fail(404, "compare task not found")
@router.post("/model-compare/{task_id}/load")
async def model_compare_load(task_id: str) -> dict[str, Any]:
try:
task = get_platform_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,
}
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}}))
except KeyError:
raise fail(404, "compare task not found")
@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": []}}))
except KeyError:
raise fail(404, "compare task not found")
@router.post("/model-compare/{task_id}/start-model")
async def model_compare_start_model(task_id: str, payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
return ok({"pid": 45001, "port": payload.get("port") or 18001, "task_id": task_id})
@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})
@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]}"})
@router.post("/model-chat/batch")
async def model_chat_batch(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
return ok({"responses": [], "request": payload})
@router.post("/model-chat/local/chat")
async def model_chat_local(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
return ok({"response": "local chat adapter is not connected yet", "request": payload})
@router.post("/model-chat/local/preload")
async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
return ok({"loaded": True, "request": payload})
@router.post("/model-chat/trained/preload")
async def model_chat_trained_preload(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
return ok({"loaded": True, "request": payload})
@router.get("/compute/nodes")
async def compute_nodes() -> dict[str, Any]:
return ok(get_platform_store().compute_nodes())
@router.get("/compute/nodes/{node_id}")
async def compute_node_detail(node_id: str) -> dict[str, Any]:
node = next((item for item in get_platform_store().compute_nodes() if item["id"] == node_id), None)
if not node:
raise fail(404, "compute node not found")
return ok(node)
@router.post("/compute/nodes")
async def create_compute_node(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
try:
return ok(get_platform_store().create_compute_node(payload))
except KeyError as exc:
raise fail(400, f"missing field: {exc}")
except ValueError as exc:
raise fail(400, str(exc))
@router.put("/compute/nodes/{node_id}")
@@ -377,11 +846,47 @@ async def update_compute_node(node_id: str, payload: dict[str, Any] = Body(...))
return ok(get_platform_store().update_compute_node(node_id, payload))
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]:
return ok({"node_id": node_id, "success": True, "latency_ms": 12})
store = get_platform_store()
node = next((item for item in store.compute_nodes() if item["id"] == node_id), None)
if not node:
raise fail(404, "compute node not found")
client = ComputeNodeClient(node["api_base_url"])
try:
result = await client.test_connection()
store.replace_node_gpus(node_id, result["gpus"])
updated = store.update_compute_node_health(node_id, result["health"], True)
return ok(
{
"node_id": node_id,
"success": True,
"latency_ms": result["latency_ms"],
"gpu_count": len(result["gpus"]),
"health": updated["health_detail"],
}
)
except Exception as exc: # noqa: BLE001 - return the connection error for node maintenance
updated = store.update_compute_node_health(node_id, {}, False, str(exc))
return ok(
{
"node_id": node_id,
"success": False,
"latency_ms": 0,
"gpu_count": updated.get("gpu_count", 0),
"error": str(exc),
"health": updated["health_detail"],
}
)
@router.post("/compute/nodes/{node_id}/health-check")
async def health_check_compute_node(node_id: str) -> dict[str, Any]:
return await test_compute_node(node_id)
@router.post("/compute/nodes/{node_id}/enable")
@@ -404,6 +909,38 @@ async def compute_node_replicas(node_id: str) -> dict[str, Any]:
return ok(get_platform_store().replicas(node_id))
@router.get("/compute/nodes/{node_id}/engines")
async def compute_node_engines(node_id: str) -> dict[str, Any]:
node = next((item for item in get_platform_store().compute_nodes() if item["id"] == node_id), None)
if not node:
raise fail(404, "compute node not found")
health = node.get("health_detail") or {}
live_error = ""
try:
health = await ComputeNodeClient(node["api_base_url"]).health()
except Exception as exc: # noqa: BLE001 - stored health is enough for offline node detail
live_error = str(exc)
capabilities = health.get("capabilities") or node.get("capabilities") or []
return ok(
{
"node_id": node_id,
"items": [
{
"engine": "llama_factory",
"display_name": "LLaMA-Factory",
"status": "available" if "llama_factory" in capabilities else "unknown",
"version": health.get("llama_factory_version") or "",
"home": health.get("llama_factory_home") or "",
"home_exists": bool(health.get("llama_factory_home_exists")),
"capabilities": capabilities,
"execution_mode": health.get("execution_mode") or "",
"last_error": live_error,
}
],
}
)
@router.get("/compute/gpus")
async def compute_gpus() -> dict[str, Any]:
return ok(get_platform_store().gpus())
@@ -414,10 +951,107 @@ async def compute_queue() -> dict[str, Any]:
return ok(get_platform_store().queue())
@router.get("/compute/jobs/{job_id}")
async def compute_job_detail(job_id: str) -> dict[str, Any]:
task = _task_for_compute_job(job_id)
if not task:
raise fail(404, "compute job not found")
node = _node_for_task(task)
if not node:
raise fail(404, "compute node not found")
return ok(await ComputeNodeClient(node["api_base_url"]).get_job(job_id))
@router.post("/compute/jobs/{job_id}/stop")
async def compute_job_stop(job_id: str) -> dict[str, Any]:
task = _task_for_compute_job(job_id)
if not task:
raise fail(404, "compute job not found")
node = _node_for_task(task)
if not node:
raise fail(404, "compute node not found")
job = await ComputeNodeClient(node["api_base_url"]).stop_job(job_id)
get_platform_store().apply_compute_job(task["id"], job)
return ok(job)
@router.get("/compute/jobs/{job_id}/logs")
async def compute_job_logs(
job_id: str,
tail_lines: int | None = Query(default=200, ge=1, le=5000),
offset: int | None = Query(default=None, ge=0),
limit: int | None = Query(default=None, ge=1, le=5000),
) -> dict[str, Any]:
task = _task_for_compute_job(job_id)
if not task:
raise fail(404, "compute job not found")
node = _node_for_task(task)
if not node:
raise fail(404, "compute node not found")
return ok(await ComputeNodeClient(node["api_base_url"]).job_logs(job_id, tail_lines, offset, limit))
@router.post("/compute/jobs/{job_id}/retry")
async def compute_job_retry(job_id: str, payload: dict[str, Any] | None = Body(default=None)) -> dict[str, Any]:
store = get_platform_store()
payload = payload or {}
task = _task_for_compute_job(job_id)
if not task:
raise fail(404, "compute job not found")
return await retry_fine_tune(task["id"], payload)
@router.post("/compute/jobs/{job_id}/priority")
async def compute_job_priority(job_id: str, payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
task = _task_for_compute_job(job_id)
if not task:
raise fail(404, "compute job not found")
priority = str(payload.get("priority") or "normal")
return ok(get_platform_store().update_task_priority(task["id"], priority))
@router.post("/internal/compute-sync/jobs/poll")
async def poll_compute_jobs() -> dict[str, Any]:
return ok(await poll_compute_jobs_once())
@router.post("/internal/compute-sync/resources")
async def create_compute_sync(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
sync_id = get_platform_store().create_sync_job(payload.get("target_node_id", "node_01"), payload)
return ok(get_platform_store().sync_job(sync_id))
store = get_platform_store()
node_id = payload.get("target_node_id") or payload.get("target_compute_node_id")
if not node_id:
raise fail(400, "target_node_id is required")
node = next((item for item in store.compute_nodes() if item["id"] == node_id), None)
if not node:
raise fail(404, "compute node not found")
sync_id = store.create_sync_job(node_id, payload)
replicas = []
failures = []
resources = payload.get("resources") or []
for resource in resources:
if not resource.get("source_path"):
continue
try:
result = await ComputeNodeClient(node["api_base_url"]).import_local_file(
{
"source_path": resource["source_path"],
"target_relative_path": resource.get("target_relative_path"),
"resource_type": resource.get("resource_type"),
"resource_id": resource.get("resource_id"),
}
)
replicas.append(
store.upsert_resource_replica(
node_id,
str(resource.get("resource_type") or "file"),
str(resource.get("resource_id") or result["id"]),
result["local_path"],
)
)
except Exception as exc: # noqa: BLE001 - collect per-resource failures
failures.append({"resource_id": str(resource.get("resource_id")), "error": str(exc)})
store.update_sync_job(sync_id, "failed" if failures else "completed", 100 if not failures else 99, completed=True)
return ok({"sync": store.sync_job(sync_id), "replicas": replicas, "failed": failures})
@router.get("/internal/compute-sync/resources/{sync_id}")

2
backend/app/core/config.py
View File

@@ -28,6 +28,8 @@ class Settings:
compute_mode: str = os.getenv("COMPUTE_MODE", "real")
compute_status_sync_mode: str = os.getenv("COMPUTE_STATUS_SYNC_MODE", "polling")
compute_poll_interval_seconds: int = _int_env("COMPUTE_POLL_INTERVAL_SECONDS", 3)
compute_request_timeout_seconds: int = _int_env("COMPUTE_REQUEST_TIMEOUT_SECONDS", 5)
compute_service_token: str = os.getenv("COMPUTE_SERVICE_TOKEN", "")
log_level: str = os.getenv("LOG_LEVEL", "INFO")
log_dir: str = os.getenv("LOG_DIR", "./logs")
log_file_prefix: str = os.getenv("LOG_FILE_PREFIX", "backend")

589
backend/app/db/platform_store.py
View File

@@ -53,6 +53,13 @@ def json_dumps(value: Any) -> str:
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
def safe_float(value: Any, default: float = 0) -> float:
try:
return float(str(value).replace("[N/A]", "").strip() or default)
except (TypeError, ValueError):
return default
def new_id(prefix: str) -> str:
return f"{prefix}_{uuid.uuid4().hex[:12]}"
@@ -178,12 +185,33 @@ class PlatformStore:
schema_path = Path(__file__).with_name("sql") / "001_platform_runtime.sql"
with self.connect() as conn:
conn.executescript(schema_path.read_text(encoding="utf-8"))
columns = conn.execute(
"SELECT column_name FROM information_schema.columns WHERE table_name='users'"
).fetchall()
column_names = {row["column_name"] for row in columns}
if "password" in column_names and "password_hash" not in column_names:
user_columns = self._column_names(conn, "users")
if "password" in user_columns and "password_hash" not in user_columns:
conn.execute("ALTER TABLE users RENAME COLUMN password TO password_hash")
self._ensure_columns(
conn,
"compute_nodes",
{
"api_version": "TEXT NOT NULL DEFAULT 'v1'",
"capabilities": "TEXT NOT NULL DEFAULT '[]'",
"description": "TEXT",
},
)
self._ensure_columns(conn, "gpus", {"last_seen_at": "TEXT"})
self._ensure_columns(conn, "fine_tune_tasks", {"compute_job_id": "TEXT"})
def _column_names(self, conn: PgConnection, table_name: str) -> set[str]:
columns = conn.execute(
"SELECT column_name FROM information_schema.columns WHERE table_name=?",
(table_name,),
).fetchall()
return {row["column_name"] for row in columns}
def _ensure_columns(self, conn: PgConnection, table_name: str, columns: dict[str, str]) -> None:
existing = self._column_names(conn, table_name)
for column, definition in columns.items():
if column not in existing:
conn.execute(f"ALTER TABLE {table_name} ADD COLUMN {column} {definition}")
def ensure_seed_data(self) -> None:
with self.connect() as conn:
@@ -436,7 +464,7 @@ class PlatformStore:
utcnow(),
),
)
return self.model(model_id)
return dict(conn.execute("SELECT * FROM models WHERE id=?", (model_id,)).fetchone())
def update_model(self, model_id: str, payload: dict[str, Any]) -> dict[str, Any]:
current = self.model(model_id)
@@ -461,7 +489,7 @@ class PlatformStore:
model_id,
),
)
return self.model(model_id)
return dict(conn.execute("SELECT * FROM models WHERE id=?", (model_id,)).fetchone())
def delete_model(self, model_id: str) -> None:
with self.connect() as conn:
@@ -481,6 +509,10 @@ class PlatformStore:
for row in rows
]
def delete_trained_model(self, model_id: str) -> None:
with self.connect() as conn:
conn.execute("DELETE FROM trained_models WHERE id=? OR name=?", (model_id, model_id))
def datasets(self) -> list[dict[str, Any]]:
with self.connect() as conn:
rows = conn.execute("SELECT * FROM datasets ORDER BY create_time DESC").fetchall()
@@ -641,6 +673,26 @@ class PlatformStore:
conn.execute("UPDATE dataset_files SET active_version_id=? WHERE id=?", (version_id, file_id))
return {"version": version, "content": row["content"]}
def delete_file_version(self, file_id: str, version_id: str) -> dict[str, Any]:
with self.connect() as conn:
row = conn.execute("SELECT * FROM dataset_files WHERE id=?", (file_id,)).fetchone()
if not row:
raise KeyError(file_id)
versions = json_loads(row["versions"], [])
if row["active_version_id"] == version_id:
raise ValueError("active dataset version cannot be deleted")
if len(versions) <= 1:
raise ValueError("last dataset version cannot be deleted")
next_versions = [item for item in versions if item["id"] != version_id]
if len(next_versions) == len(versions):
raise KeyError(version_id)
conn.execute("UPDATE dataset_files SET versions=? WHERE id=?", (json_dumps(next_versions), file_id))
return {
"versions": next_versions,
"active_version_id": row["active_version_id"],
"next_version_number": max(item.get("version", 0) for item in next_versions) + 1,
}
def tasks(self) -> list[dict[str, Any]]:
self.refresh_runtime_state()
with self.connect() as conn:
@@ -668,6 +720,8 @@ class PlatformStore:
"train_duration": self._duration(row["start_time"], row["completed_at"]) if row["start_time"] else "",
"compute_node_id": row["compute_node_id"],
"sync_job_id": row["sync_job_id"],
"compute_job_id": row.get("compute_job_id"),
"completed_at": row.get("completed_at"),
}
)
return payload
@@ -689,6 +743,7 @@ class PlatformStore:
"status": "pending",
"train_type": payload.get("train_type", "SFT"),
"train_method": payload.get("train_method", "lora"),
"engine": payload.get("engine", payload.get("training_engine", "llama_factory")),
"template": payload.get("template", "qwen"),
"base_model": base_model,
"train_dataset_id": train_dataset_id,
@@ -751,7 +806,7 @@ class PlatformStore:
"""
UPDATE fine_tune_tasks
SET payload=?, status='syncing', progress=8, process_id=?, start_time=?,
compute_node_id=?, gpus=?, sync_job_id=?
compute_node_id=?, gpus=?, sync_job_id=?, compute_job_id=NULL
WHERE id=?
""",
(
@@ -766,9 +821,140 @@ class PlatformStore:
)
return self.task(task_id)
def stop_task(self, task_id: str) -> dict[str, Any]:
def reset_task_for_retry(self, task_id: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
current = self.task(task_id)
override = payload or {}
merged = {
**current,
**override,
"id": task_id,
"status": "pending",
"progress": 0,
"process_id": None,
"compute_job_id": None,
}
for runtime_key in ["failure_reason", "log_file", "artifacts"]:
merged.pop(runtime_key, None)
with self.connect() as conn:
conn.execute(
"""
UPDATE fine_tune_tasks
SET payload=?, status='pending', progress=0, process_id=NULL, start_time=NULL,
completed_at=NULL, compute_node_id=NULL, gpus=?, sync_job_id=NULL, compute_job_id=NULL
WHERE id=?
""",
(json_dumps(merged), json_dumps(merged.get("gpus", [])), task_id),
)
return self.task(task_id)
def update_task_priority(self, task_id: str, priority: str) -> dict[str, Any]:
current = self.task(task_id)
priority = priority if priority in {"low", "normal", "high", "urgent"} else "normal"
merged = {**current, "priority": priority}
with self.connect() as conn:
conn.execute("UPDATE fine_tune_tasks SET payload=? WHERE id=?", (json_dumps(merged), task_id))
return self.task(task_id)
def prepare_compute_job_payload(self, task_id: str, payload: dict[str, Any] | None = None) -> tuple[dict[str, Any], dict[str, Any]]:
task = self.task(task_id)
task.update({"status": "failed", "progress": min(task.get("progress", 0), 99)})
merged = {**task, **(payload or {}), "id": task_id}
node = self.schedule_node(merged)
selected_gpus = merged.get("gpus") or [0]
return node, self._compute_job_payload_from_task_node(merged, node, selected_gpus)
def _compute_job_payload_from_task_node(
self,
task: dict[str, Any],
node: dict[str, Any],
selected_gpus: list[int] | list[Any] | None = None,
) -> dict[str, Any]:
with self.connect() as conn:
model = conn.execute("SELECT * FROM models WHERE id=?", (task.get("base_model"),)).fetchone()
dataset = conn.execute("SELECT * FROM datasets WHERE id=?", (task.get("train_dataset_id"),)).fetchone()
model_path = (model and model.get("path")) or task.get("base_model")
dataset_name = task.get("dataset") or (dataset and dataset.get("name")) or task.get("train_dataset_id")
health_detail = node.get("health_detail") or {}
dataset_root = str(health_detail.get("dataset_root") or f"{node['data_root'].rstrip('/')}/datasets")
output_root = str(health_detail.get("output_root") or f"{node['data_root'].rstrip('/')}/outputs")
output_dir = task.get("output_dir") or f"{output_root.rstrip('/')}/{task['name']}"
return {
**task,
"id": task["id"],
"name": task["name"],
"base_model": model_path,
"model_name_or_path": model_path,
"dataset": dataset_name,
"dataset_dir": dataset_root,
"output_dir": output_dir,
"gpus": selected_gpus or task.get("gpus") or [0],
"compute_node_id": node["id"],
"compute_node_code": node["code"],
}
def build_compute_job_payload(self, task_id: str) -> tuple[dict[str, Any], dict[str, Any]]:
task = self.task(task_id)
node = next((item for item in self.compute_nodes() if item["id"] == task.get("compute_node_id")), None)
if not node:
raise RuntimeError("compute node not found")
return node, self._compute_job_payload_from_task_node(task, node, task.get("gpus") or [0])
def apply_compute_job(self, task_id: str, job: dict[str, Any]) -> dict[str, Any]:
status_map = {
"queued": "queued",
"running": "running",
"completed": "completed",
"failed": "failed",
"stopped": "stopped",
}
current = self.task(task_id)
status = status_map.get(str(job.get("status")), str(job.get("status") or current["status"]))
progress = int(job.get("progress", current.get("progress", 0)) or 0)
payload = {
**current,
"status": status,
"progress": progress,
"process_id": job.get("pid") or current.get("process_id"),
"compute_job_id": job.get("id") or current.get("compute_job_id"),
"output_dir": job.get("output_dir") or current.get("output_dir"),
"log_file": job.get("log_file") or current.get("log_file"),
"artifacts": job.get("artifacts") or current.get("artifacts") or [],
}
if status == "failed":
payload["failure_reason"] = job.get("error") or job.get("message") or current.get("failure_reason") or "compute job failed"
elif status in {"queued", "running", "completed"}:
payload.pop("failure_reason", None)
completed_at = utcnow() if status in {"completed", "failed", "stopped"} and not current.get("completed_at") else None
with self.connect() as conn:
conn.execute(
"""
UPDATE fine_tune_tasks
SET payload=?, status=?, progress=?, process_id=?, compute_job_id=?, completed_at=COALESCE(?, completed_at)
WHERE id=?
""",
(
json_dumps(payload),
status,
progress,
payload.get("process_id"),
payload.get("compute_job_id"),
completed_at,
task_id,
),
)
if status == "completed":
self._ensure_trained_model(conn, payload)
return self.task(task_id)
def running_compute_tasks(self) -> list[dict[str, Any]]:
return [
task
for task in self.tasks()
if task.get("compute_job_id") and task.get("compute_node_id") and task["status"] in {"syncing", "queued", "running"}
]
def mark_task_failed(self, task_id: str, reason: str) -> dict[str, Any]:
task = self.task(task_id)
task.update({"status": "failed", "progress": min(task.get("progress", 0), 99), "failure_reason": reason})
with self.connect() as conn:
conn.execute(
"UPDATE fine_tune_tasks SET status='failed', payload=?, completed_at=? WHERE id=?",
@@ -776,10 +962,179 @@ class PlatformStore:
)
return self.task(task_id)
def stop_task(self, task_id: str, status: str = "stopped") -> dict[str, Any]:
task = self.task(task_id)
task.update({"status": status, "progress": min(task.get("progress", 0), 99)})
with self.connect() as conn:
conn.execute(
"UPDATE fine_tune_tasks SET status=?, payload=?, completed_at=? WHERE id=?",
(status, json_dumps(task), utcnow(), task_id),
)
return self.task(task_id)
def delete_task(self, task_id: str) -> None:
with self.connect() as conn:
conn.execute("DELETE FROM fine_tune_tasks WHERE id=?", (task_id,))
def _json_payload_row(self, row: PgRow) -> dict[str, Any]:
payload = json_loads(row["payload"], {})
payload.update({"id": row["id"], "status": row.get("status"), "create_time": row["create_time"]})
return payload
def eval_tasks(self) -> list[dict[str, Any]]:
with self.connect() as conn:
rows = conn.execute("SELECT * FROM eval_tasks ORDER BY create_time DESC").fetchall()
return [self._json_payload_row(row) for row in rows]
def eval_task(self, task_id: str) -> dict[str, Any]:
with self.connect() as conn:
row = conn.execute("SELECT * FROM eval_tasks WHERE id=?", (task_id,)).fetchone()
if not row:
raise KeyError(task_id)
payload = self._json_payload_row(row)
payload.setdefault("sample_count", 0)
payload.setdefault("completed_count", 0)
payload.setdefault("passed_count", 0)
payload.setdefault("overall_score", payload.get("score") or 0)
payload.setdefault("overall_score_max", 100)
payload.setdefault("overall_evaluation", "")
payload.setdefault("improvement_suggestions", [])
payload.setdefault("dimension_summary", [])
payload.setdefault("samples", [])
return payload
def create_eval_task(self, payload: dict[str, Any]) -> dict[str, Any]:
task_id = str(payload.get("id") or payload.get("task_id") or new_id("eval"))
name = str(payload.get("eval_task_name") or payload.get("name") or f"eval-{task_id[-6:]}")
status = str(payload.get("status") or "pending")
now = payload.get("create_time") or utcnow()
data = {
**payload,
"id": task_id,
"eval_task_name": name,
"status": status,
"create_time": now,
"metric": payload.get("metric") or "custom",
}
with self.connect() as conn:
model = conn.execute("SELECT name FROM models WHERE id=?", (str(payload.get("model_id")),)).fetchone()
dataset = conn.execute("SELECT name FROM datasets WHERE id=?", (str(payload.get("dataset_id")),)).fetchone()
if model:
data.setdefault("model_name", model["name"])
if dataset:
data.setdefault("dataset", dataset["name"])
conn.execute(
"INSERT INTO eval_tasks (id, name, payload, status, create_time) VALUES (?, ?, ?, ?, ?)",
(task_id, name, json_dumps(data), status, now),
)
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 dimensions(self) -> list[dict[str, Any]]:
with self.connect() as conn:
rows = conn.execute("SELECT * FROM eval_dimensions ORDER BY create_time DESC").fetchall()
return [
{
**json_loads(row["payload"], {}),
"id": row["id"],
"name": row["name"],
"is_active": bool(row["is_active"]),
"is_default": bool(row["is_default"]),
"create_time": row["create_time"],
}
for row in rows
]
def dimension(self, dimension_id: str) -> dict[str, Any]:
with self.connect() as conn:
row = conn.execute("SELECT * FROM eval_dimensions WHERE id=?", (dimension_id,)).fetchone()
if not row:
raise KeyError(dimension_id)
return {
**json_loads(row["payload"], {}),
"id": row["id"],
"name": row["name"],
"is_active": bool(row["is_active"]),
"is_default": bool(row["is_default"]),
"create_time": row["create_time"],
}
def create_dimension(self, payload: dict[str, Any]) -> dict[str, Any]:
dimension_id = str(payload.get("id") or new_id("dim"))
name = str(payload.get("name") or f"dimension-{dimension_id[-6:]}")
now = payload.get("create_time") or utcnow()
data = {**payload, "id": dimension_id, "name": name, "create_time": now}
with self.connect() as conn:
conn.execute(
"INSERT INTO eval_dimensions (id, name, payload, is_active, is_default, create_time) VALUES (?, ?, ?, ?, ?, ?)",
(dimension_id, name, json_dumps(data), 1 if data.get("is_active", True) else 0, 1 if data.get("is_default") else 0, now),
)
return self.dimension(dimension_id)
def update_dimension(self, dimension_id: str, payload: dict[str, Any]) -> dict[str, Any]:
current = self.dimension(dimension_id)
merged = {**current, **payload, "id": dimension_id}
with self.connect() as conn:
conn.execute(
"UPDATE eval_dimensions SET name=?, payload=?, is_active=?, is_default=? WHERE id=?",
(
merged["name"],
json_dumps(merged),
1 if merged.get("is_active", True) else 0,
1 if merged.get("is_default") else 0,
dimension_id,
),
)
return self.dimension(dimension_id)
def delete_dimension(self, dimension_id: str) -> None:
with self.connect() as conn:
conn.execute("DELETE FROM eval_dimensions WHERE id=?", (dimension_id,))
def compare_tasks(self) -> list[dict[str, Any]]:
with self.connect() as conn:
rows = conn.execute("SELECT * FROM compare_tasks ORDER BY create_time DESC").fetchall()
return [self._json_payload_row(row) for row in rows]
def compare_task(self, task_id: str) -> dict[str, Any]:
with self.connect() as conn:
row = conn.execute("SELECT * FROM compare_tasks WHERE id=?", (task_id,)).fetchone()
if not row:
raise KeyError(task_id)
return self._json_payload_row(row)
def create_compare_task(self, payload: dict[str, Any]) -> dict[str, Any]:
task_id = str(payload.get("id") or new_id("cmp"))
name = str(payload.get("name") or payload.get("model_name") or f"compare-{task_id[-6:]}")
status = str(payload.get("status") or "pending")
now = payload.get("create_time") or utcnow()
data = {**payload, "id": task_id, "name": name, "model_name": payload.get("model_name") or name, "status": status, "create_time": now}
data.setdefault("load_status", json_dumps({"loaded_models": []}))
with self.connect() as conn:
conn.execute(
"INSERT INTO compare_tasks (id, name, payload, status, create_time) VALUES (?, ?, ?, ?, ?)",
(task_id, name, json_dumps(data), status, now),
)
return self.compare_task(task_id)
def update_compare_task(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]:
current = self.compare_task(task_id)
merged = {**current, **payload, "id": task_id}
status = str(merged.get("status") or current.get("status") or "pending")
with self.connect() as conn:
conn.execute(
"UPDATE compare_tasks SET name=?, payload=?, status=? WHERE id=?",
(merged.get("name") or merged.get("model_name") or task_id, json_dumps(merged), status, task_id),
)
return self.compare_task(task_id)
def delete_compare_task(self, task_id: str) -> None:
with self.connect() as conn:
conn.execute("DELETE FROM compare_tasks WHERE id=?", (task_id,))
def schedule_node(self, payload: dict[str, Any]) -> dict[str, Any]:
requested = payload.get("requested_node_id") or payload.get("compute_node_id")
nodes = self.compute_nodes()
@@ -793,7 +1148,20 @@ class PlatformStore:
if selected:
return selected
if not candidates:
raise RuntimeError("no available compute node")
if not nodes:
raise RuntimeError("no available compute node: no compute node configured")
reasons = []
for node in nodes:
if not node["enabled"]:
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']}"
else:
reason = "not selected"
reasons.append(f"{node['code']}({reason})")
raise RuntimeError(f"no available compute node: {', '.join(reasons)}")
return sorted(candidates, key=lambda n: (-n["scheduler_weight"], n["current_running_jobs"], n["code"]))[0]
def create_sync_job(self, node_id: str, task: dict[str, Any]) -> str:
@@ -809,7 +1177,8 @@ class PlatformStore:
sync_id,
node_id,
json_dumps(
[
task.get("resources")
or [
{"resource_type": "model", "resource_id": task.get("base_model")},
{"resource_type": "dataset", "resource_id": task.get("train_dataset_id")},
]
@@ -819,6 +1188,46 @@ class PlatformStore:
)
return sync_id
def update_sync_job(self, sync_id: str, status: str, progress: int, completed: bool = False) -> dict[str, Any]:
with self.connect() as conn:
conn.execute(
"UPDATE resource_sync_jobs SET status=?, progress=?, completed_at=COALESCE(?, completed_at) WHERE id=?",
(status, progress, utcnow() if completed else None, sync_id),
)
return self.sync_job(sync_id)
def upsert_resource_replica(
self,
node_id: str,
resource_type: str,
resource_id: str,
local_path: str,
status: str = "available",
sync_status: str = "synced",
) -> dict[str, Any]:
with self.connect() as conn:
row = conn.execute(
"SELECT * FROM resource_replicas WHERE node_id=? AND resource_type=? AND resource_id=?",
(node_id, resource_type, resource_id),
).fetchone()
if row:
conn.execute(
"UPDATE resource_replicas SET local_path=?, status=?, sync_status=? WHERE id=?",
(local_path, status, sync_status, row["id"]),
)
replica_id = row["id"]
else:
replica_id = new_id("replica")
conn.execute(
"""
INSERT INTO resource_replicas
(id, node_id, resource_type, resource_id, local_path, status, sync_status, create_time)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(replica_id, node_id, resource_type, resource_id, local_path, status, sync_status, utcnow()),
)
return dict(conn.execute("SELECT * FROM resource_replicas WHERE id=?", (replica_id,)).fetchone())
def progress(self, task_id: str) -> dict[str, Any]:
task = self.task(task_id)
status = task.get("status", "pending")
@@ -829,6 +1238,7 @@ class PlatformStore:
"running": "training with LLaMA-Factory",
"completed": "training completed",
"failed": "training stopped",
"stopped": "training stopped",
}
progress = int(task.get("progress", 0) or 0)
eta = "--" if status in {"completed", "failed"} else f"{max(1, math.ceil((100 - progress) / 10))} min"
@@ -853,23 +1263,62 @@ class PlatformStore:
**dict(row),
"enabled": bool(row["enabled"]),
"tags": json_loads(row["tags"], []),
"capabilities": json_loads(row.get("capabilities"), []),
"health_detail": json_loads(row["health_detail"], {}),
"current_running_jobs": running_map.get(row["id"], 0),
}
for row in rows
]
def _normalize_tags(self, value: Any) -> list[str]:
if isinstance(value, str):
parts = value.replace("，", ",").split(",")
return [item.strip() for item in parts if item.strip()]
if isinstance(value, list):
return [str(item).strip() for item in value if str(item).strip()]
return []
def _normalize_compute_node_payload(self, payload: dict[str, Any], current: dict[str, Any] | None = None) -> dict[str, Any]:
merged = {**(current or {}), **payload}
api_base_url = str(merged.get("api_base_url") or "").rstrip("/")
if not api_base_url:
raise ValueError("api_base_url is required")
file_gateway_url = str(merged.get("file_gateway_url") or api_base_url).rstrip("/")
weight = max(0, min(1000, int(merged.get("scheduler_weight", 100))))
max_jobs = max(1, int(merged.get("max_parallel_jobs", 1)))
return {
**merged,
"code": str(merged.get("code") or "").strip(),
"name": str(merged.get("name") or merged.get("code") or "").strip(),
"api_base_url": api_base_url,
"file_gateway_url": file_gateway_url,
"enabled": bool(merged.get("enabled", True)),
"scheduler_status": str(merged.get("scheduler_status") or "offline"),
"scheduler_weight": weight,
"tags": self._normalize_tags(merged.get("tags")),
"gpu_count": max(0, int(merged.get("gpu_count", 0) or 0)),
"max_parallel_jobs": max_jobs,
"data_root": str(merged.get("data_root") or "/data/yg-ft"),
"model_root": str(merged.get("model_root") or "/data/yg-ft/models"),
"log_root": str(merged.get("log_root") or "/opt/yg-ft/logs/training"),
"api_version": str(merged.get("api_version") or "v1"),
"capabilities": merged.get("capabilities") or [],
"description": merged.get("description") or "",
"health_detail": merged.get("health_detail") or {"status": "registered"},
}
def update_compute_node(self, node_id: str, payload: dict[str, Any]) -> dict[str, Any]:
current = next((n for n in self.compute_nodes() if n["id"] == node_id), None)
if not current:
raise KeyError(node_id)
merged = {**current, **payload}
merged = self._normalize_compute_node_payload(payload, current)
with self.connect() as conn:
conn.execute(
"""
UPDATE compute_nodes
SET name=?, api_base_url=?, file_gateway_url=?, enabled=?, scheduler_status=?,
scheduler_weight=?, tags=?, max_parallel_jobs=?, last_health_check_at=?
scheduler_weight=?, tags=?, max_parallel_jobs=?, data_root=?, model_root=?, log_root=?,
api_version=?, capabilities=?, description=?, last_health_check_at=?, health_detail=?
WHERE id=?
""",
(
@@ -881,46 +1330,116 @@ class PlatformStore:
merged["scheduler_weight"],
json_dumps(merged["tags"]),
merged["max_parallel_jobs"],
utcnow(),
merged["data_root"],
merged["model_root"],
merged["log_root"],
merged["api_version"],
json_dumps(merged["capabilities"]),
merged["description"],
payload.get("last_health_check_at") or current.get("last_health_check_at"),
json_dumps(merged["health_detail"]),
node_id,
),
)
return next(n for n in self.compute_nodes() if n["id"] == node_id)
def create_compute_node(self, payload: dict[str, Any]) -> dict[str, Any]:
payload = self._normalize_compute_node_payload(payload)
if not payload["code"]:
raise ValueError("code is required")
node_id = payload.get("id") or new_id("node")
now = utcnow()
tags = payload.get("tags") or []
with self.connect() as conn:
conn.execute(
"""
INSERT INTO compute_nodes
(id, code, name, api_base_url, file_gateway_url, enabled, scheduler_status,
scheduler_weight, tags, gpu_count, current_running_jobs, max_parallel_jobs,
data_root, model_root, log_root, last_health_check_at, health_detail)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?)
data_root, model_root, log_root, api_version, capabilities, description,
last_health_check_at, health_detail)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
node_id,
payload["code"],
payload.get("name") or payload["code"],
payload["name"] or payload["code"],
payload["api_base_url"],
payload.get("file_gateway_url") or payload["api_base_url"],
1 if payload.get("enabled", True) else 0,
payload.get("scheduler_status", "offline"),
int(payload.get("scheduler_weight", 100)),
json_dumps(tags),
int(payload.get("gpu_count", 0)),
int(payload.get("max_parallel_jobs", 1)),
payload.get("data_root", "/data/yg-ft"),
payload.get("model_root", "/models"),
payload.get("log_root", "/data/yg-ft/training-logs"),
payload["file_gateway_url"],
1 if payload["enabled"] else 0,
payload["scheduler_status"],
payload["scheduler_weight"],
json_dumps(payload["tags"]),
payload["gpu_count"],
payload["max_parallel_jobs"],
payload["data_root"],
payload["model_root"],
payload["log_root"],
payload["api_version"],
json_dumps(payload["capabilities"]),
payload["description"],
now,
json_dumps(payload.get("health_detail") or {"status": "registered"}),
json_dumps(payload["health_detail"]),
),
)
return next(node for node in self.compute_nodes() if node["id"] == 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:
raise KeyError(node_id)
status = "online" if success and current.get("enabled") else "offline"
if current.get("scheduler_status") == "draining" and success:
status = "draining"
detail = {
**(current.get("health_detail") or {}),
**health,
"status": "ok" if success else "failed",
"last_error": error or "",
"checked_at": utcnow(),
}
return self.update_compute_node(
node_id,
{
"scheduler_status": status,
"last_health_check_at": detail["checked_at"],
"health_detail": detail,
"data_root": health.get("data_root") or current.get("data_root"),
"api_version": str(health.get("api_version") or current.get("api_version") or "v1"),
"capabilities": health.get("capabilities") or current.get("capabilities") or [],
},
)
def replace_node_gpus(self, node_id: str, gpus: list[dict[str, Any]]) -> None:
now = utcnow()
with self.connect() as conn:
conn.execute("DELETE FROM gpus WHERE node_id=?", (node_id,))
for index, item in enumerate(gpus):
gpu_index = int(item.get("gpu_index", item.get("id", index)) or 0)
memory_total = safe_float(item.get("memory_total_gb") or item.get("memory_total"))
if not memory_total and item.get("memory_total_mb") is not None:
memory_total = round(safe_float(item.get("memory_total_mb")) / 1024, 2)
power_limit = safe_float(item.get("power_limit_w") or item.get("power_limit"))
temperature = int(safe_float(item.get("temperature") or item.get("base_temperature"), 35))
conn.execute(
"""
INSERT INTO gpus
(id, node_id, gpu_index, uuid, name, memory_total_gb, power_limit_w, base_temperature, last_seen_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
f"{node_id}_gpu_{gpu_index}",
node_id,
gpu_index,
str(item.get("uuid") or f"{node_id}-GPU-{gpu_index}"),
str(item.get("name") or "Unknown GPU"),
memory_total or 0,
power_limit or 0,
temperature,
now,
),
)
conn.execute("UPDATE compute_nodes SET gpu_count=? WHERE id=?", (len(gpus), node_id))
def gpus(self) -> list[dict[str, Any]]:
self.refresh_runtime_state()
with self.connect() as conn:
@@ -951,6 +1470,7 @@ class PlatformStore:
reserved = task is not None and task.get("status") in {"syncing", "queued"}
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)
items.append(
{
"id": row["gpu_index"],
@@ -961,8 +1481,8 @@ class PlatformStore:
"uuid": row["uuid"],
"gpu_percent": gpu_percent,
"memory_used_gb": memory_used,
"memory_total_gb": row["memory_total_gb"],
"memory_percent": round(memory_used / row["memory_total_gb"] * 100, 1),
"memory_total_gb": memory_total,
"memory_percent": round(memory_used / memory_total * 100, 1) if memory_total else 0,
"temperature": row["base_temperature"] + (21 if busy else 6 if reserved else 0),
"power_w": round(row["power_limit_w"] * (0.7 if busy else 0.25 if reserved else 0.08), 1),
"power_limit_w": row["power_limit_w"],
@@ -1034,12 +1554,14 @@ class PlatformStore:
}
def queue(self) -> list[dict[str, Any]]:
return [
priority_score = {"urgent": 3, "high": 2, "normal": 1, "low": 0}
items = [
{
"id": task["id"],
"name": task["name"],
"status": task["status"],
"progress": task.get("progress", 0),
"priority": task.get("priority", "normal"),
"compute_node_id": task.get("compute_node_id"),
"gpus": task.get("gpus", []),
"create_time": task.get("create_time"),
@@ -1047,6 +1569,7 @@ class PlatformStore:
for task in self.tasks()
if task["status"] in {"pending", "syncing", "queued", "running"}
]
return sorted(items, key=lambda item: (-priority_score.get(item["priority"], 1), item["create_time"]), reverse=False)
def replicas(self, node_id: str) -> list[dict[str, Any]]:
with self.connect() as conn:

42
backend/app/db/sql/001_platform_runtime.sql
View File

@@ -76,6 +76,9 @@ CREATE TABLE IF NOT EXISTS compute_nodes (
data_root TEXT NOT NULL,
model_root TEXT NOT NULL,
log_root TEXT NOT NULL,
api_version TEXT NOT NULL DEFAULT 'v1',
capabilities TEXT NOT NULL DEFAULT '[]',
description TEXT,
last_health_check_at TEXT,
health_detail TEXT NOT NULL
);
@@ -88,7 +91,8 @@ CREATE TABLE IF NOT EXISTS gpus (
name TEXT NOT NULL,
memory_total_gb DOUBLE PRECISION NOT NULL,
power_limit_w DOUBLE PRECISION NOT NULL,
base_temperature INTEGER NOT NULL
base_temperature INTEGER NOT NULL,
last_seen_at TEXT
);
CREATE TABLE IF NOT EXISTS fine_tune_tasks (
@@ -103,7 +107,8 @@ CREATE TABLE IF NOT EXISTS fine_tune_tasks (
completed_at TEXT,
compute_node_id TEXT REFERENCES compute_nodes(id) ON DELETE SET NULL,
gpus TEXT NOT NULL,
sync_job_id TEXT
sync_job_id TEXT,
compute_job_id TEXT
);
CREATE TABLE IF NOT EXISTS resource_replicas (
@@ -127,7 +132,40 @@ CREATE TABLE IF NOT EXISTS resource_sync_jobs (
completed_at TEXT
);
CREATE TABLE IF NOT EXISTS eval_tasks (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
payload TEXT NOT NULL,
status TEXT NOT NULL,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS eval_dimensions (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
payload TEXT NOT NULL,
is_active INTEGER NOT NULL DEFAULT 1,
is_default INTEGER NOT NULL DEFAULT 0,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS compare_tasks (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
payload TEXT NOT NULL,
status TEXT NOT NULL,
create_time TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_fine_tune_status ON fine_tune_tasks(status);
CREATE INDEX IF NOT EXISTS idx_fine_tune_compute_job ON fine_tune_tasks(compute_job_id);
CREATE INDEX IF NOT EXISTS idx_fine_tune_compute_node_status ON fine_tune_tasks(compute_node_id, status);
CREATE INDEX IF NOT EXISTS idx_dataset_files_dataset ON dataset_files(dataset_id);
CREATE INDEX IF NOT EXISTS idx_gpus_node ON gpus(node_id);
CREATE UNIQUE INDEX IF NOT EXISTS uq_gpus_node_index ON gpus(node_id, gpu_index);
CREATE INDEX IF NOT EXISTS idx_replicas_resource ON resource_replicas(resource_type, resource_id);
CREATE UNIQUE INDEX IF NOT EXISTS uq_replicas_node_resource ON resource_replicas(node_id, resource_type, resource_id);
CREATE INDEX IF NOT EXISTS idx_sync_jobs_node_status ON resource_sync_jobs(target_node_id, status);
CREATE INDEX IF NOT EXISTS idx_eval_tasks_status ON eval_tasks(status);
CREATE INDEX IF NOT EXISTS idx_eval_dimensions_active ON eval_dimensions(is_active);
CREATE INDEX IF NOT EXISTS idx_compare_tasks_status ON compare_tasks(status);

17
backend/app/main.py
View File

@@ -1,9 +1,13 @@
import asyncio
from contextlib import suppress
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.v1.router import api_router
from app.core.config import get_settings
from app.core.logging import configure_logging, setup_request_logging
from app.workers.compute_poller import run_compute_poller
def create_app() -> FastAPI:
@@ -20,6 +24,19 @@ def create_app() -> FastAPI:
)
setup_request_logging(app)
app.include_router(api_router, prefix=settings.route_prefix)
@app.on_event("startup")
async def start_workers() -> None:
app.state.compute_poller_task = asyncio.create_task(run_compute_poller())
@app.on_event("shutdown")
async def stop_workers() -> None:
task = getattr(app.state, "compute_poller_task", None)
if task:
task.cancel()
with suppress(asyncio.CancelledError):
await task
return app

206
backend/app/modules/compute_gateway/client.py Normal file
View File

@@ -0,0 +1,206 @@
from __future__ import annotations
import time
from typing import Any
from urllib.parse import urljoin
import httpx
from app.core.config import get_settings
def _join_url(base_url: str, path: str) -> str:
return urljoin(base_url.rstrip("/") + "/", path.lstrip("/"))
def _unwrap_items(payload: Any) -> list[dict[str, Any]]:
if isinstance(payload, list):
return [item for item in payload if isinstance(item, dict)]
if isinstance(payload, dict):
data = payload.get("data")
if isinstance(data, dict) and isinstance(data.get("items"), list):
return [item for item in data["items"] if isinstance(item, dict)]
if isinstance(payload.get("items"), list):
return [item for item in payload["items"] if isinstance(item, dict)]
if isinstance(data, list):
return [item for item in data if isinstance(item, dict)]
return []
def _unwrap_dict(payload: Any) -> dict[str, Any]:
if isinstance(payload, dict) and isinstance(payload.get("data"), dict):
return payload["data"]
return payload if isinstance(payload, dict) else {}
class ComputeNodeClient:
"""Application-side client for one compute node.
The client accepts both current YG Compute API responses and common
wrapper shapes such as `{code,message,data}` to make future engine/node
adapters less brittle.
"""
def __init__(self, api_base_url: str, token: str | None = None, timeout: float | None = None) -> None:
settings = get_settings()
self.api_base_url = api_base_url.rstrip("/")
self.token = token or settings.compute_service_token
self.timeout = timeout or settings.compute_request_timeout_seconds
self.route_prefix = settings.route_prefix.rstrip("/") or "/modelTF"
def headers(self) -> dict[str, str]:
if not self.token:
return {}
return {"X-Compute-Token": self.token}
async def test_connection(self) -> dict[str, Any]:
started = time.perf_counter()
health = await self.health()
gpus = await self.gpus()
return {
"success": True,
"latency_ms": int((time.perf_counter() - started) * 1000),
"health": health,
"gpus": gpus,
}
async def health(self) -> dict[str, Any]:
paths = [f"{self.route_prefix}/v1/compute/health", f"{self.route_prefix}/health", "/health"]
last_error = ""
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client:
for path in paths:
try:
response = await client.get(_join_url(self.api_base_url, path))
response.raise_for_status()
return _unwrap_dict(response.json())
except Exception as exc: # noqa: BLE001 - keep endpoint compatibility fallback broad
last_error = str(exc)
raise RuntimeError(last_error or "compute health check failed")
async def gpus(self) -> list[dict[str, Any]]:
paths = [
f"{self.route_prefix}/compute/resources/gpus",
f"{self.route_prefix}/v1/compute/resources/gpus",
"/compute/resources/gpus",
]
last_error = ""
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client:
for path in paths:
try:
response = await client.get(_join_url(self.api_base_url, path))
response.raise_for_status()
return _unwrap_items(response.json())
except Exception as exc: # noqa: BLE001
last_error = str(exc)
raise RuntimeError(last_error or "compute gpu discovery failed")
async def create_job(self, payload: dict[str, Any]) -> dict[str, Any]:
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client:
response = await client.post(_join_url(self.api_base_url, f"{self.route_prefix}/compute/jobs"), json=payload)
response.raise_for_status()
return _unwrap_dict(response.json())
async def preview_job(self, payload: dict[str, Any]) -> dict[str, Any]:
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client:
response = await client.post(
_join_url(self.api_base_url, f"{self.route_prefix}/compute/jobs/preview"),
json=payload,
)
response.raise_for_status()
return _unwrap_dict(response.json())
async def validate_job(self, payload: dict[str, Any]) -> dict[str, Any]:
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client:
response = await client.post(
_join_url(self.api_base_url, f"{self.route_prefix}/compute/jobs/validate"),
json=payload,
)
response.raise_for_status()
return _unwrap_dict(response.json())
async def check_paths(self, paths: list[dict[str, Any]]) -> dict[str, Any]:
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client:
response = await client.post(
_join_url(self.api_base_url, f"{self.route_prefix}/compute/files/check-paths"),
json={"paths": paths},
)
response.raise_for_status()
return _unwrap_dict(response.json())
async def list_files(
self,
root: str = "data",
relative_path: str = "",
directories_only: bool = False,
) -> dict[str, Any]:
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client:
response = await client.get(
_join_url(self.api_base_url, f"{self.route_prefix}/compute/files/list"),
params={"root": root, "relative_path": relative_path, "directories_only": directories_only},
)
response.raise_for_status()
return _unwrap_dict(response.json())
async def get_job(self, job_id: str) -> dict[str, Any]:
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client:
response = await client.get(_join_url(self.api_base_url, f"{self.route_prefix}/compute/jobs/{job_id}"))
response.raise_for_status()
return _unwrap_dict(response.json())
async def stop_job(self, job_id: str) -> dict[str, Any]:
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client:
response = await client.post(_join_url(self.api_base_url, f"{self.route_prefix}/compute/jobs/{job_id}/stop"))
response.raise_for_status()
return _unwrap_dict(response.json())
async def job_logs(
self,
job_id: str,
tail_lines: int | None = None,
offset: int | None = None,
limit: int | None = None,
) -> dict[str, Any]:
params = {
key: value
for key, value in {"tail_lines": tail_lines, "offset": offset, "limit": limit}.items()
if value is not None
}
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client:
response = await client.get(
_join_url(self.api_base_url, f"{self.route_prefix}/compute/jobs/{job_id}/logs"),
params=params,
)
response.raise_for_status()
return _unwrap_dict(response.json())
async def import_local_file(self, payload: dict[str, Any]) -> dict[str, Any]:
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client:
response = await client.post(
_join_url(self.api_base_url, f"{self.route_prefix}/compute/files/import-local"),
json=payload,
)
response.raise_for_status()
return _unwrap_dict(response.json())
async def upload_file(
self,
filename: str,
content: bytes,
target_relative_path: str,
resource_type: str | None = None,
resource_id: str | None = None,
) -> dict[str, Any]:
data = {
"target_relative_path": target_relative_path,
"resource_type": resource_type or "",
"resource_id": resource_id or "",
}
files = {"file": (filename, content)}
async with httpx.AsyncClient(timeout=max(self.timeout, 60), headers=self.headers()) as client:
response = await client.post(
_join_url(self.api_base_url, f"{self.route_prefix}/compute/files/upload"),
data=data,
files=files,
)
response.raise_for_status()
return _unwrap_dict(response.json())

27
backend/app/modules/compute_gateway/sync.py Normal file
View File

@@ -0,0 +1,27 @@
from __future__ import annotations
from typing import Any
from app.db.platform_store import get_platform_store
from app.modules.compute_gateway.client import ComputeNodeClient
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 poll_compute_jobs_once() -> dict[str, Any]:
store = get_platform_store()
synced: list[dict[str, Any]] = []
failed: list[dict[str, str]] = []
for task in store.running_compute_tasks():
node = _node_for_task(task)
if not node:
failed.append({"task_id": task["id"], "error": "compute node not found"})
continue
try:
job = await ComputeNodeClient(node["api_base_url"]).get_job(task["compute_job_id"])
synced.append(store.apply_compute_job(task["id"], job))
except Exception as exc: # noqa: BLE001 - keep polling other jobs
failed.append({"task_id": task["id"], "error": str(exc)})
return {"synced": len(synced), "failed": failed, "items": synced}

31
backend/app/workers/compute_poller.py Normal file
View File

@@ -0,0 +1,31 @@
from __future__ import annotations
import asyncio
from app.core.config import get_settings
from app.core.logging import get_logger
from app.modules.compute_gateway.sync import poll_compute_jobs_once
logger = get_logger(__name__)
async def run_compute_poller() -> None:
settings = get_settings()
if settings.compute_mode == "simulator" or settings.compute_status_sync_mode != "polling":
logger.info("compute poller disabled", extra={"compute_mode": settings.compute_mode})
return
interval = max(3, settings.compute_poll_interval_seconds)
logger.info("compute poller started", extra={"interval_seconds": interval})
while True:
try:
result = await poll_compute_jobs_once()
if result["synced"] or result["failed"]:
logger.info("compute jobs polled", extra={"result": result})
except asyncio.CancelledError:
logger.info("compute poller stopped")
raise
except Exception as exc: # noqa: BLE001 - keep background polling alive
logger.exception("compute poller failed", extra={"error": str(exc)})
await asyncio.sleep(interval)