Codex Release Bot
125 lines
3.8 KiB
Python
125 lines
3.8 KiB
Python
"""Central Nexus Agent version (SSOT) and semver-style comparison."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
_AGENT_PY = Path(__file__).resolve().parents[2] / "web" / "agent" / "agent.py"
|
|
_AGENT_VERSION_CONST_RE = re.compile(r'^AGENT_VERSION\s*=\s*"([^"]+)"', re.MULTILINE)
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_central_agent_version() -> str:
|
|
"""Read AGENT_VERSION constant from web/agent/agent.py (heartbeat SSOT)."""
|
|
text = _AGENT_PY.read_text(encoding="utf-8")
|
|
match = _AGENT_VERSION_CONST_RE.search(text)
|
|
if not match:
|
|
raise RuntimeError(f"AGENT_VERSION not found in {_AGENT_PY}")
|
|
return match.group(1)
|
|
|
|
|
|
def version_tuple(version: str) -> tuple[int, ...]:
|
|
"""Parse dotted numeric prefix, e.g. 2.0.0 -> (2, 0, 0)."""
|
|
version = (version or "").strip()
|
|
if not version:
|
|
return ()
|
|
parts: list[int] = []
|
|
for segment in re.split(r"[.-]", version):
|
|
if segment.isdigit():
|
|
parts.append(int(segment))
|
|
else:
|
|
break
|
|
return tuple(parts)
|
|
|
|
|
|
def is_version_lower(remote: str, central: str) -> bool:
|
|
"""True when remote is strictly older than central."""
|
|
remote_t = version_tuple(remote)
|
|
central_t = version_tuple(central)
|
|
if not remote_t:
|
|
return True
|
|
if not central_t:
|
|
return False
|
|
return remote_t < central_t
|
|
|
|
|
|
def is_version_at_least(remote: str, central: str) -> bool:
|
|
"""True when remote >= central (equal counts as skip)."""
|
|
remote_t = version_tuple(remote)
|
|
central_t = version_tuple(central)
|
|
if not remote_t:
|
|
return False
|
|
if not central_t:
|
|
return bool(remote_t)
|
|
return remote_t >= central_t
|
|
|
|
|
|
def agent_is_installed(
|
|
*,
|
|
agent_version_db: str | None,
|
|
heartbeat: dict[str, str] | None = None,
|
|
) -> bool:
|
|
"""True when Agent has reported a version or heartbeat (not merely a DB API key)."""
|
|
if heartbeat:
|
|
if (heartbeat.get("agent_version") or "").strip():
|
|
return True
|
|
if heartbeat.get("last_heartbeat"):
|
|
return True
|
|
return bool((agent_version_db or "").strip())
|
|
|
|
|
|
def compute_agent_guidance(
|
|
*,
|
|
agent_version_db: str | None,
|
|
heartbeat: dict[str, str] | None = None,
|
|
central: str | None = None,
|
|
) -> dict[str, str | None]:
|
|
"""UI hints: install / upgrade / offline / ok."""
|
|
central_v = central or get_central_agent_version()
|
|
live_ver = ""
|
|
if heartbeat:
|
|
live_ver = (heartbeat.get("agent_version") or "").strip()
|
|
if not live_ver:
|
|
live_ver = (agent_version_db or "").strip()
|
|
|
|
has_heartbeat = bool(heartbeat and heartbeat.get("last_heartbeat"))
|
|
installed = agent_is_installed(agent_version_db=agent_version_db, heartbeat=heartbeat)
|
|
|
|
base: dict[str, str | None] = {
|
|
"central_agent_version": central_v,
|
|
"agent_action": "ok",
|
|
"agent_action_message": None,
|
|
}
|
|
|
|
if not installed:
|
|
return {
|
|
**base,
|
|
"agent_action": "install",
|
|
"agent_action_message": (
|
|
f"尚未安装 Agent（主站版本 {central_v}）。"
|
|
"安装后可自动上报 CPU/内存/磁盘与在线状态；未安装时仅能通过 SSH 手动探活。"
|
|
),
|
|
}
|
|
|
|
if live_ver and is_version_lower(live_ver, central_v):
|
|
return {
|
|
**base,
|
|
"agent_action": "upgrade",
|
|
"agent_action_message": (
|
|
f"当前 Agent {live_ver}，低于主站 {central_v}，建议升级以获取最新心跳与探针能力。"
|
|
),
|
|
}
|
|
|
|
if not has_heartbeat:
|
|
return {
|
|
**base,
|
|
"agent_action": "offline",
|
|
"agent_action_message": (
|
|
"Agent 已登记但暂无心跳，请检查子机 nexus-agent 服务或点击「诊断」。"
|
|
),
|
|
}
|
|
|
|
return base
|