sysadminandClaude Opus 4.8
95f77bb648
Adds read-only /runtime and /api/runtime routes showing active profile, identity, config model, git sync vs remote master, shell health, workflow/schema hashes, and stale-runtime warnings with #420 guidance. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
320 lines
10 KiB
Python
320 lines
10 KiB
Python
"""Read-only MCP/runtime health snapshot for the web UI (#430)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
from urllib.parse import urlparse
|
|
|
|
import gitea_config
|
|
import reconciler_profile
|
|
from gitea_auth import REMOTES, api_request, get_auth_header, get_profile, gitea_url
|
|
|
|
from native_mcp_preference import shell_health_status
|
|
|
|
from webui.project_registry import ProjectRecord, load_registry
|
|
|
|
_RESTART_DOCS = "docs/llm-workflow-runbooks.md#issue-420-mcp-hot-reload"
|
|
_HOT_RELOAD_ISSUE = "#420"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FileHash:
|
|
label: str
|
|
path: str
|
|
sha256: str | None
|
|
error: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RuntimeSnapshot:
|
|
project_id: str
|
|
repo_root: str
|
|
remote: str
|
|
host: str
|
|
profile_name: str
|
|
role_kind: str
|
|
config_model: str
|
|
profile_mode: str
|
|
profile_source: str
|
|
authenticated_username: str | None
|
|
identity_error: str | None
|
|
repo_sha: str | None
|
|
remote_master_sha: str | None
|
|
commits_behind_master: int | None
|
|
stale_runtime_warning: str | None
|
|
shell_health: dict[str, Any]
|
|
workflow_hashes: tuple[FileHash, ...]
|
|
schema_hashes: tuple[FileHash, ...]
|
|
restart_guidance: str
|
|
fetch_error: str | None = None
|
|
|
|
|
|
def _repo_root() -> Path:
|
|
override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
|
|
if override:
|
|
return Path(override).resolve()
|
|
return Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def _host_from_url(remote_host: str) -> str:
|
|
parsed = urlparse(remote_host.strip())
|
|
return parsed.netloc or remote_host.strip().rstrip("/")
|
|
|
|
|
|
def _remote_name_for_host(host: str) -> str:
|
|
for name, profile in REMOTES.items():
|
|
if profile.get("host") == host:
|
|
return name
|
|
return "custom"
|
|
|
|
|
|
def _config_model(config: dict | None) -> str:
|
|
if config is None:
|
|
return "env-only"
|
|
version = config.get("version")
|
|
if version == 1:
|
|
return "v1"
|
|
if version == 2:
|
|
if "contexts" in config or config.get("shape") == "contexts":
|
|
return "v2-contexts"
|
|
return "v2-environments"
|
|
return f"unknown-version-{version}"
|
|
|
|
|
|
def _profile_source() -> str:
|
|
if os.environ.get("GITEA_PROFILE_NAME"):
|
|
return "env var"
|
|
if gitea_config.selected_profile_name():
|
|
return "config file profile"
|
|
return "default"
|
|
|
|
|
|
def _role_kind(allowed: list[str], forbidden: list[str]) -> str:
|
|
if reconciler_profile.is_reconciler_profile(allowed, forbidden):
|
|
return "reconciler"
|
|
|
|
def can(op: str) -> bool:
|
|
return gitea_config.check_operation(op, allowed, forbidden)[0]
|
|
|
|
review = can("gitea.pr.approve") or can("gitea.pr.merge")
|
|
author = can("gitea.pr.create") or can("gitea.branch.push")
|
|
reconciler = can("gitea.pr.close") and not review and not author
|
|
if review and author:
|
|
return "mixed"
|
|
if review:
|
|
return "reviewer"
|
|
if reconciler:
|
|
return "reconciler"
|
|
if author:
|
|
return "author"
|
|
return "limited"
|
|
|
|
|
|
def _sha256_file(path: Path) -> FileHash:
|
|
label = path.name
|
|
try:
|
|
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
return FileHash(label=label, path=str(path), sha256=digest)
|
|
except OSError as exc:
|
|
return FileHash(label=label, path=str(path), sha256=None, error=str(exc))
|
|
|
|
|
|
def _run_git(repo: Path, *args: str) -> str | None:
|
|
try:
|
|
completed = subprocess.run(
|
|
["git", *args],
|
|
cwd=repo,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return completed.stdout.strip() or None
|
|
except (OSError, subprocess.CalledProcessError):
|
|
return None
|
|
|
|
|
|
def _git_sync_status(repo: Path, remote: str, branch: str) -> tuple[str | None, str | None, int | None]:
|
|
local_sha = _run_git(repo, "rev-parse", "HEAD")
|
|
remote_sha = _run_git(repo, "rev-parse", f"{remote}/{branch}")
|
|
behind = None
|
|
if local_sha and remote_sha and local_sha != remote_sha:
|
|
count = _run_git(repo, "rev-list", "--count", f"HEAD..{remote}/{branch}")
|
|
if count is not None and count.isdigit():
|
|
behind = int(count)
|
|
return local_sha, remote_sha, behind
|
|
|
|
|
|
def _resolve_username(host: str) -> tuple[str | None, str | None]:
|
|
auth = get_auth_header(host)
|
|
if not auth:
|
|
return None, f"Gitea credentials unavailable for {host}"
|
|
try:
|
|
data = api_request("GET", gitea_url(host, "/api/v1/user"), auth)
|
|
except Exception as exc: # noqa: BLE001 — operator-visible identity errors
|
|
return None, f"Identity lookup failed: {exc}"
|
|
username = (data or {}).get("login")
|
|
if not username:
|
|
return None, "Authenticated identity could not be resolved"
|
|
return str(username), None
|
|
|
|
|
|
def _stale_warning(commits_behind: int | None) -> str | None:
|
|
if commits_behind is None or commits_behind <= 0:
|
|
return None
|
|
return (
|
|
f"Local checkout is {commits_behind} commit(s) behind remote master. "
|
|
"Merged safety-gate changes may require an MCP server restart or profile "
|
|
f"reload — see {_HOT_RELOAD_ISSUE} / {_RESTART_DOCS}."
|
|
)
|
|
|
|
|
|
def load_runtime_snapshot(
|
|
project_id: str | None = None,
|
|
*,
|
|
resolve_username: Callable[[str], tuple[str | None, str | None]] | None = None,
|
|
git_sync: Callable[[Path, str, str], tuple[str | None, str | None, int | None]] | None = None,
|
|
) -> RuntimeSnapshot:
|
|
"""Build a read-only runtime health snapshot for the default registry project."""
|
|
registry = load_registry()
|
|
project: ProjectRecord | None = None
|
|
if project_id:
|
|
for entry in registry.projects:
|
|
if entry.id == project_id:
|
|
project = entry
|
|
break
|
|
else:
|
|
project = registry.projects[0] if registry.projects else None
|
|
|
|
if project is None:
|
|
return RuntimeSnapshot(
|
|
project_id=project_id or "",
|
|
repo_root=str(_repo_root()),
|
|
remote="",
|
|
host="",
|
|
profile_name="",
|
|
role_kind="",
|
|
config_model="",
|
|
profile_mode="",
|
|
profile_source="",
|
|
authenticated_username=None,
|
|
identity_error="project not found in registry",
|
|
repo_sha=None,
|
|
remote_master_sha=None,
|
|
commits_behind_master=None,
|
|
stale_runtime_warning=None,
|
|
shell_health={},
|
|
workflow_hashes=(),
|
|
schema_hashes=(),
|
|
restart_guidance=_RESTART_DOCS,
|
|
fetch_error="project not found in registry",
|
|
)
|
|
|
|
repo = _repo_root()
|
|
host = _host_from_url(project.remote_host)
|
|
remote = _remote_name_for_host(host)
|
|
profile = get_profile()
|
|
allowed = profile.get("allowed_operations") or []
|
|
forbidden = profile.get("forbidden_operations") or []
|
|
config = None
|
|
try:
|
|
config = gitea_config.load_config()
|
|
except gitea_config.ConfigError as exc:
|
|
return RuntimeSnapshot(
|
|
project_id=project.id,
|
|
repo_root=str(repo),
|
|
remote=remote,
|
|
host=host,
|
|
profile_name=profile.get("profile_name") or "",
|
|
role_kind=_role_kind(allowed, forbidden),
|
|
config_model="config-error",
|
|
profile_mode="unknown",
|
|
profile_source=_profile_source(),
|
|
authenticated_username=None,
|
|
identity_error=str(exc),
|
|
repo_sha=None,
|
|
remote_master_sha=None,
|
|
commits_behind_master=None,
|
|
stale_runtime_warning=None,
|
|
shell_health=shell_health_status(),
|
|
workflow_hashes=(),
|
|
schema_hashes=(),
|
|
restart_guidance=_RESTART_DOCS,
|
|
fetch_error=str(exc),
|
|
)
|
|
|
|
identity_fn = resolve_username or _resolve_username
|
|
username, identity_error = identity_fn(host)
|
|
|
|
git_fn = git_sync or _git_sync_status
|
|
remote_ref = "prgs" if remote == "prgs" else "origin"
|
|
local_sha, remote_sha, behind = git_fn(repo, remote_ref, project.default_branch)
|
|
|
|
workflow_hashes = tuple(
|
|
_sha256_file(repo / rel_path)
|
|
for rel_path in project.workflow_paths.values()
|
|
)
|
|
schema_hashes = tuple(
|
|
_sha256_file(repo / rel_path)
|
|
for rel_path in project.schema_paths.values()
|
|
)
|
|
|
|
switching = gitea_config.is_runtime_switching_enabled()
|
|
return RuntimeSnapshot(
|
|
project_id=project.id,
|
|
repo_root=str(repo),
|
|
remote=remote,
|
|
host=host,
|
|
profile_name=str(profile.get("profile_name") or ""),
|
|
role_kind=_role_kind(allowed, forbidden),
|
|
config_model=_config_model(config),
|
|
profile_mode="dynamic-profile" if switching else "static-profile",
|
|
profile_source=_profile_source(),
|
|
authenticated_username=username,
|
|
identity_error=identity_error,
|
|
repo_sha=local_sha,
|
|
remote_master_sha=remote_sha,
|
|
commits_behind_master=behind,
|
|
stale_runtime_warning=_stale_warning(behind),
|
|
shell_health=shell_health_status(),
|
|
workflow_hashes=workflow_hashes,
|
|
schema_hashes=schema_hashes,
|
|
restart_guidance=_RESTART_DOCS,
|
|
)
|
|
|
|
|
|
def snapshot_to_dict(snapshot: RuntimeSnapshot) -> dict[str, Any]:
|
|
def _hash_row(item: FileHash) -> dict[str, Any]:
|
|
return {
|
|
"label": item.label,
|
|
"path": item.path,
|
|
"sha256": item.sha256,
|
|
"error": item.error,
|
|
}
|
|
|
|
return {
|
|
"project_id": snapshot.project_id,
|
|
"repo_root": snapshot.repo_root,
|
|
"remote": snapshot.remote,
|
|
"host": snapshot.host,
|
|
"profile_name": snapshot.profile_name,
|
|
"role_kind": snapshot.role_kind,
|
|
"config_model": snapshot.config_model,
|
|
"profile_mode": snapshot.profile_mode,
|
|
"profile_source": snapshot.profile_source,
|
|
"authenticated_username": snapshot.authenticated_username,
|
|
"identity_error": snapshot.identity_error,
|
|
"repo_sha": snapshot.repo_sha,
|
|
"remote_master_sha": snapshot.remote_master_sha,
|
|
"commits_behind_master": snapshot.commits_behind_master,
|
|
"stale_runtime_warning": snapshot.stale_runtime_warning,
|
|
"shell_health": snapshot.shell_health,
|
|
"workflow_hashes": [_hash_row(item) for item in snapshot.workflow_hashes],
|
|
"schema_hashes": [_hash_row(item) for item in snapshot.schema_hashes],
|
|
"restart_guidance": snapshot.restart_guidance,
|
|
"fetch_error": snapshot.fetch_error,
|
|
} |