docs/webui-local-dev.md

@@ -43,7 +43,8 @@ Optional environment variables:
 | `/api/projects` | JSON registry export |
 | `/prompts` | Prompt library with per-prompt copy buttons (#428) |
 | `/api/prompts` | JSON prompt export with workflow hashes |
-| `/runtime` | Stub — MCP runtime health (#430) |
+| `/runtime` | MCP runtime health and stale detection (#430) |
+| `/api/runtime` | JSON runtime health export |
 | `/audit` | Stub — report audit paste (#431) |
 | `/worktrees` | Stub — hygiene dashboard (#432) |
 | `/leases` | Lease and collision visibility (#433) |
@@ -91,8 +92,17 @@ in-progress claim inventory (#268), reviewer PR lease comments when present
 and duplicate local branches per issue. Links to collision-history backend
 issues (#267, #268, #400, #407) are included. No lease acquire/release from UI.
 
+## Runtime health (#430)
+
+`/runtime` surfaces read-only MCP/runtime diagnostics for the default registry
+project: active profile and role kind, authenticated identity (when credentials
+are available), config model/mode, local vs remote `master` SHA sync, shell
+health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
+checkout is behind merged safety-gate changes. Restart guidance links to #420;
+no tokens or MCP restart actions are exposed.
+
 ## Tests
 
 ```bash
-pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_lease_visibility.py -q
+pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_lease_visibility.py tests/test_webui_runtime_health.py -q
 ```

tests/test_webui_runtime_health.py

@@ -0,0 +1,88 @@
+"""Tests for web UI runtime health dashboard (#430)."""
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from starlette.testclient import TestClient
+
+from webui.app import create_app
+from webui.runtime_health import _role_kind, load_runtime_snapshot, snapshot_to_dict
+
+
+def _mock_identity(_host: str):
+    return "jcwalker3", None
+
+
+def _mock_git_sync(_repo: Path, _remote: str, _branch: str):
+    return "localsha123456", "remoteshaabcdef", 2
+
+
+class TestRuntimeHealthLoader(unittest.TestCase):
+    def test_role_kind_author(self):
+        allowed = ["gitea.pr.create", "gitea.branch.push", "gitea.read"]
+        forbidden = ["gitea.pr.merge", "gitea.pr.approve"]
+        self.assertEqual(_role_kind(allowed, forbidden), "author")
+
+    def test_snapshot_with_mocks(self):
+        snapshot = load_runtime_snapshot(
+            resolve_username=_mock_identity,
+            git_sync=_mock_git_sync,
+        )
+        self.assertEqual(snapshot.project_id, "gitea-tools")
+        self.assertEqual(snapshot.authenticated_username, "jcwalker3")
+        self.assertEqual(snapshot.commits_behind_master, 2)
+        self.assertIsNotNone(snapshot.stale_runtime_warning)
+        self.assertGreaterEqual(len(snapshot.workflow_hashes), 3)
+        self.assertGreaterEqual(len(snapshot.schema_hashes), 2)
+        self.assertIn("shell_use_allowed", snapshot.shell_health)
+
+    def test_snapshot_dict_export(self):
+        snapshot = load_runtime_snapshot(
+            resolve_username=_mock_identity,
+            git_sync=_mock_git_sync,
+        )
+        data = snapshot_to_dict(snapshot)
+        self.assertEqual(data["authenticated_username"], "jcwalker3")
+        self.assertEqual(data["commits_behind_master"], 2)
+        self.assertTrue(data["stale_runtime_warning"])
+
+
+class TestRuntimeRoutes(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+        self.snapshot = load_runtime_snapshot(
+            resolve_username=_mock_identity,
+            git_sync=_mock_git_sync,
+        )
+        self._patch = mock.patch(
+            "webui.app.load_runtime_snapshot",
+            return_value=self.snapshot,
+        )
+        self._patch.start()
+
+    def tearDown(self):
+        self._patch.stop()
+
+    def test_runtime_page_renders_profile_and_warning(self):
+        response = self.client.get("/runtime")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Runtime health", response.text)
+        self.assertIn("Active profile", response.text)
+        self.assertIn("Stale runtime warning", response.text)
+        self.assertIn("Workflow hashes", response.text)
+        self.assertNotIn("child issue", response.text.lower())
+
+    def test_api_runtime_json(self):
+        response = self.client.get("/api/runtime")
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertEqual(data["project_id"], "gitea-tools")
+        self.assertEqual(data["role_kind"], self.snapshot.role_kind)
+        self.assertEqual(data["commits_behind_master"], 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_skeleton.py

@@ -29,8 +29,13 @@ class TestWebuiSkeleton(unittest.TestCase):
         self.assertIn("Operator console", response.text)
         self.assertIn("Read-only MVP", response.text)
 
+    def test_runtime_is_implemented(self):
+        response = self.client.get("/runtime")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Runtime health", response.text)
+
     def test_route_stubs_render(self):
-        for path in ("/runtime", "/audit"):
+        for path in ("/audit",):
             with self.subTest(path=path):
                 response = self.client.get(path)
                 self.assertEqual(response.status_code, 200)

webui/app.py

@@ -16,8 +16,10 @@ from webui.prompt_library import find_prompt, library_to_dict
 from webui.prompt_views import render_prompt_detail, render_prompts_page
 from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict
 from webui.lease_views import render_leases_page
-from webui.queue_loader import load_queue_snapshot, snapshot_to_dict
+from webui.queue_loader import load_queue_snapshot, snapshot_to_dict as queue_snapshot_to_dict
 from webui.queue_views import render_queue_page
+from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
+from webui.runtime_views import render_runtime_page
 
 _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 
@@ -63,7 +65,7 @@ async def queue(_request: Request) -> HTMLResponse:
 
 
 async def api_queue(_request: Request) -> JSONResponse:
-    return JSONResponse(snapshot_to_dict(load_queue_snapshot()))
+    return JSONResponse(queue_snapshot_to_dict(load_queue_snapshot()))
 
 
 async def projects(_request: Request) -> HTMLResponse:
@@ -122,10 +124,12 @@ async def api_prompts(_request: Request) -> JSONResponse:
 
 
 async def runtime(_request: Request) -> HTMLResponse:
-    return _stub_page(
-        "Runtime",
-        "Runtime health will report MCP profile, preflight, and stale-server signals.",
-    )
+    snapshot = load_runtime_snapshot()
+    return HTMLResponse(render_page(title="Runtime", body_html=render_runtime_page(snapshot)))
+
+
+async def api_runtime(_request: Request) -> JSONResponse:
+    return JSONResponse(runtime_snapshot_to_dict(load_runtime_snapshot()))
 
 
 async def audit(_request: Request) -> HTMLResponse:
@@ -176,6 +180,7 @@ def create_app() -> Starlette:
             Route("/prompts/{prompt_id}", prompt_detail, methods=["GET"]),
             Route("/api/prompts", api_prompts, methods=["GET"]),
             Route("/runtime", runtime, methods=["GET"]),
+            Route("/api/runtime", api_runtime, methods=["GET"]),
             Route("/audit", audit, methods=["GET"]),
             Route("/worktrees", worktrees, methods=["GET"]),
             Route("/leases", leases, methods=["GET"]),

webui/lease_loader.py

@@ -11,8 +11,8 @@ from urllib.parse import urlparse
 
 from gitea_auth import api_fetch_page, get_auth_header, repo_api_url
 from issue_claim_heartbeat import build_claim_inventory
-from merged_cleanup_reconcile import read_issue_lock
 from issue_lock_provenance import ISSUE_LOCK_FILE
+from merged_cleanup_reconcile import read_issue_lock
 
 from webui.project_registry import ProjectRecord, load_registry
 from webui.queue_loader import _extract_linked_issue, _fetch_issues, _fetch_prs

webui/runtime_health.py

@@ -0,0 +1,320 @@
+"""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,
+    }

webui/runtime_views.py

@@ -0,0 +1,92 @@
+"""HTML views for MCP runtime health (#430)."""
+
+from __future__ import annotations
+
+import html
+
+from webui.runtime_health import FileHash, RuntimeSnapshot
+
+
+def _hash_rows(items: tuple[FileHash, ...]) -> str:
+    if not items:
+        return "<p class='muted'>No hashes available.</p>"
+    rows = []
+    for item in items:
+        digest = item.sha256 or item.error or "unavailable"
+        rows.append(
+            "<tr>"
+            f"<td>{html.escape(item.label)}</td>"
+            f"<td><code>{html.escape(item.path)}</code></td>"
+            f"<td><code>{html.escape(digest[:16] if item.sha256 else digest)}</code></td>"
+            "</tr>"
+        )
+    return (
+        "<table class='registry'><thead><tr>"
+        "<th>Artifact</th><th>Path</th><th>SHA-256</th>"
+        "</tr></thead><tbody>"
+        f"{''.join(rows)}</tbody></table>"
+    )
+
+
+def render_runtime_page(snapshot: RuntimeSnapshot) -> str:
+    error_block = ""
+    if snapshot.fetch_error:
+        error_block = (
+            '<div class="stub"><p><strong>Runtime snapshot incomplete:</strong> '
+            f"{html.escape(snapshot.fetch_error)}</p></div>"
+        )
+
+    identity = snapshot.authenticated_username or "unresolved"
+    if snapshot.identity_error:
+        identity = f"unresolved ({snapshot.identity_error})"
+
+    stale_block = ""
+    if snapshot.stale_runtime_warning:
+        stale_block = (
+            '<div class="stub"><p><strong>Stale runtime warning:</strong> '
+            f"{html.escape(snapshot.stale_runtime_warning)}</p></div>"
+        )
+
+    shell = snapshot.shell_health or {}
+    shell_summary = (
+        f"shell_use_allowed={shell.get('shell_use_allowed')} · "
+        f"failures={shell.get('consecutive_spawn_failures')} · "
+        f"hard_stopped={shell.get('hard_stopped')}"
+    )
+
+    return (
+        "<h2>Runtime health</h2>"
+        f"<p class='meta'>Project <code>{html.escape(snapshot.project_id)}</code> · "
+        f"host <code>{html.escape(snapshot.host)}</code> · "
+        f"repo root <code>{html.escape(snapshot.repo_root)}</code></p>"
+        f"{error_block}"
+        f"{stale_block}"
+        "<h3>Active profile</h3>"
+        "<table class='detail'>"
+        f"<tr><th>Profile</th><td><code>{html.escape(snapshot.profile_name)}</code></td></tr>"
+        f"<tr><th>Role kind</th><td>{html.escape(snapshot.role_kind)}</td></tr>"
+        f"<tr><th>Identity</th><td>{html.escape(identity)}</td></tr>"
+        f"<tr><th>Config model</th><td>{html.escape(snapshot.config_model)}</td></tr>"
+        f"<tr><th>Profile mode</th><td>{html.escape(snapshot.profile_mode)} "
+        f"({html.escape(snapshot.profile_source)})</td></tr>"
+        "</table>"
+        "<h3>Server code sync</h3>"
+        "<table class='detail'>"
+        f"<tr><th>Local HEAD</th><td><code>{html.escape(snapshot.repo_sha or 'unknown')}</code></td></tr>"
+        f"<tr><th>Remote {html.escape(snapshot.remote)}/{html.escape('master')}</th>"
+        f"<td><code>{html.escape(snapshot.remote_master_sha or 'unknown')}</code></td></tr>"
+        f"<tr><th>Commits behind</th><td>{snapshot.commits_behind_master if snapshot.commits_behind_master is not None else 'unknown'}</td></tr>"
+        "</table>"
+        "<h3>Shell health</h3>"
+        f"<p class='meta'>{html.escape(shell_summary)}</p>"
+        f"<p class='muted'>{html.escape(str(shell.get('safe_next_action') or ''))}</p>"
+        "<h3>Workflow hashes</h3>"
+        f"{_hash_rows(snapshot.workflow_hashes)}"
+        "<h3>Schema hashes</h3>"
+        f"{_hash_rows(snapshot.schema_hashes)}"
+        "<h3>Restart / reload</h3>"
+        "<p class='muted'>MVP is read-only — restart MCP servers from your IDE/operator "
+        "workflow. Related issue: <code>#420</code>. Guidance: "
+        f"<code>{html.escape(snapshot.restart_guidance)}</code></p>"
+        "<p class='muted'>This page does not expose tokens or perform MCP restarts.</p>"
+    )