sysadminandClaude Opus 4.8
b2f6e9a6dc
Evolve the MVP project registry (#427) into a versioned, fail-closed project registry API for the console (Phase 1, read-only). - Add schema version 2 with project `status`, per-step onboarding `state`/`required`, optional redacted `last_seen_health`, and `remote_name`. Version 1 files stay loadable and are normalized with explicit defaults. - Serve `/api/v1/projects` and `/api/v1/projects/{project_id}` with API provenance (`api_version`, `schema_version`, `source`). `/api/projects` is retained as an unversioned Phase 1 alias. - Replace bare `ValueError` with `RegistryError`, carrying an operator `remediation` and `field_path`; invalid registries fail closed as a 500 JSON payload or a dedicated HTML error page instead of a traceback. - Reject credential-shaped keys before any DTO is built, reusing `registry_safety.is_forbidden_key` as the single source of truth shared with the worker registry (#798). - Render HTML views from `project_to_dict`, so the console and the JSON API cannot disagree about status or onboarding progress. - Document the contract in docs/webui-project-registry-api.md. Tests: registry load/validate (valid, missing project, schema validation, credential rejection) and API route coverage. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
600 lines
20 KiB
Python
600 lines
20 KiB
Python
"""Load and validate the web UI project registry (#427, evolved for #635).
|
|
|
|
Phase 1 of the console architecture ADR keeps this loader read-only. It owns
|
|
the versioned project registry contract served at ``/api/v1/projects``:
|
|
|
|
* the on-disk file carries a ``version`` (schema version 1 or 2);
|
|
* version 1 files stay loadable and are normalized with explicit defaults, so
|
|
an operator registry written for #427 keeps working;
|
|
* every validation failure raises :class:`RegistryError`, which carries an
|
|
actionable ``remediation`` string instead of leaking a traceback;
|
|
* serialization never emits credentials — credential-shaped keys are rejected
|
|
at load time, before any DTO is built.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from webui.registry_safety import is_forbidden_key
|
|
|
|
#: Version of the JSON contract served under ``/api/v1/...``.
|
|
REGISTRY_API_VERSION = "v1"
|
|
|
|
#: Schema version written by this repository's packaged registry.
|
|
CURRENT_SCHEMA_VERSION = 2
|
|
|
|
#: Schema versions this loader accepts. Version 1 is normalized on load.
|
|
SUPPORTED_SCHEMA_VERSIONS = (1, 2)
|
|
|
|
#: Lifecycle state of a registered project.
|
|
PROJECT_STATUSES = ("active", "onboarding", "paused", "archived")
|
|
_DEFAULT_PROJECT_STATUS = "active"
|
|
|
|
#: Completion state of a single onboarding step.
|
|
ONBOARDING_STATES = ("complete", "pending", "blocked", "not_applicable")
|
|
_DEFAULT_ONBOARDING_STATE = "pending"
|
|
|
|
#: Redacted, last-seen health of a project's control plane.
|
|
HEALTH_STATUSES = ("healthy", "degraded", "unreachable", "unknown")
|
|
|
|
class RegistryError(ValueError):
|
|
"""A registry file could not be loaded or failed validation.
|
|
|
|
Carries an operator-facing ``remediation`` so routes can fail closed with
|
|
an actionable message rather than a stack trace.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
*,
|
|
remediation: str,
|
|
source_path: Path | None = None,
|
|
field_path: str | None = None,
|
|
) -> None:
|
|
super().__init__(message)
|
|
self.message = message
|
|
self.remediation = remediation
|
|
self.source_path = source_path
|
|
self.field_path = field_path
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Serialize for a fail-closed JSON error response."""
|
|
return {
|
|
"error": "registry_invalid",
|
|
"detail": self.message,
|
|
"remediation": self.remediation,
|
|
"field_path": self.field_path,
|
|
"source_path": str(self.source_path) if self.source_path else None,
|
|
}
|
|
|
|
|
|
_REQUIRED_PROJECT_FIELDS = (
|
|
"id",
|
|
"repo_name",
|
|
"gitea_owner",
|
|
"remote_host",
|
|
"default_branch",
|
|
"local_checkout_path",
|
|
"profiles",
|
|
"workflow_paths",
|
|
)
|
|
|
|
_REQUIRED_PROFILE_ROLES = ("author", "reviewer", "reconciler")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OnboardingStep:
|
|
id: str
|
|
title: str
|
|
description: str
|
|
state: str = _DEFAULT_ONBOARDING_STATE
|
|
required: bool = True
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OnboardingSummary:
|
|
"""Aggregate onboarding progress for a single project."""
|
|
|
|
total: int
|
|
complete: int
|
|
pending: int
|
|
blocked: int
|
|
not_applicable: int
|
|
required_outstanding: int
|
|
onboarding_complete: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProjectHealth:
|
|
"""Redacted last-seen health. Never carries endpoints or credentials."""
|
|
|
|
status: str
|
|
checked_at: str | None
|
|
detail: str | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProjectRecord:
|
|
id: str
|
|
repo_name: str
|
|
gitea_owner: str
|
|
remote_host: str
|
|
default_branch: str
|
|
local_checkout_path: str
|
|
profiles: dict[str, str]
|
|
workflow_paths: dict[str, str]
|
|
schema_paths: dict[str, str]
|
|
onboarding_checklist: tuple[OnboardingStep, ...]
|
|
status: str = _DEFAULT_PROJECT_STATUS
|
|
remote_name: str | None = None
|
|
last_seen_health: ProjectHealth | None = None
|
|
|
|
@property
|
|
def repo_full_name(self) -> str:
|
|
return f"{self.gitea_owner}/{self.repo_name}"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProjectRegistry:
|
|
version: int
|
|
projects: tuple[ProjectRecord, ...]
|
|
source_path: Path
|
|
|
|
@property
|
|
def schema_version(self) -> int:
|
|
"""Alias of :attr:`version` — the schema version read from disk."""
|
|
return self.version
|
|
|
|
@property
|
|
def api_version(self) -> str:
|
|
return REGISTRY_API_VERSION
|
|
|
|
|
|
def default_registry_path() -> Path:
|
|
override = os.environ.get("WEBUI_PROJECT_REGISTRY", "").strip()
|
|
if override:
|
|
return Path(override).expanduser().resolve()
|
|
return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve()
|
|
|
|
|
|
def _reject_credential_keys(obj: Any, *, path: str = "", source: Path | None = None) -> None:
|
|
"""Recursive credential-key guard that reports an actionable ``field_path``.
|
|
|
|
Key *shape* is decided by :func:`webui.registry_safety.is_forbidden_key`, the
|
|
single source of truth shared with the worker registry (#798).
|
|
"""
|
|
if isinstance(obj, dict):
|
|
for key, value in obj.items():
|
|
key_path = f"{path}.{key}" if path else key
|
|
if is_forbidden_key(key):
|
|
raise RegistryError(
|
|
f"registry must not store credentials ({key_path})",
|
|
remediation=(
|
|
f"Remove the credential-shaped key '{key_path}' from the registry. "
|
|
"Tokens live in the keychain and are resolved server-side by "
|
|
"gitea_auth; the registry is redacted metadata only."
|
|
),
|
|
source_path=source,
|
|
field_path=key_path,
|
|
)
|
|
_reject_credential_keys(value, path=key_path, source=source)
|
|
elif isinstance(obj, list):
|
|
for index, item in enumerate(obj):
|
|
_reject_credential_keys(item, path=f"{path}[{index}]", source=source)
|
|
|
|
|
|
def _require_enum(
|
|
value: Any,
|
|
*,
|
|
allowed: tuple[str, ...],
|
|
field_path: str,
|
|
source: Path | None,
|
|
) -> str:
|
|
text = str(value)
|
|
if text not in allowed:
|
|
raise RegistryError(
|
|
f"{field_path} must be one of {', '.join(allowed)} (got {text!r})",
|
|
remediation=(
|
|
f"Set {field_path} to one of: {', '.join(allowed)}. "
|
|
"Unknown values fail closed so the console never renders an "
|
|
"unverified state."
|
|
),
|
|
source_path=source,
|
|
field_path=field_path,
|
|
)
|
|
return text
|
|
|
|
|
|
def _parse_onboarding(
|
|
raw: Any,
|
|
*,
|
|
project_path: str,
|
|
source: Path | None,
|
|
) -> tuple[OnboardingStep, ...]:
|
|
if raw is None:
|
|
return ()
|
|
if not isinstance(raw, list):
|
|
raise RegistryError(
|
|
f"{project_path}.onboarding_checklist must be an array",
|
|
remediation=(
|
|
f"Rewrite {project_path}.onboarding_checklist as a JSON array of "
|
|
"steps with id, title, description, and optional state."
|
|
),
|
|
source_path=source,
|
|
field_path=f"{project_path}.onboarding_checklist",
|
|
)
|
|
steps: list[OnboardingStep] = []
|
|
for index, item in enumerate(raw):
|
|
step_path = f"{project_path}.onboarding_checklist[{index}]"
|
|
if not isinstance(item, dict):
|
|
raise RegistryError(
|
|
f"{step_path} must be an object",
|
|
remediation=f"Rewrite {step_path} as an object with id, title, description.",
|
|
source_path=source,
|
|
field_path=step_path,
|
|
)
|
|
missing = [field for field in ("id", "title", "description") if field not in item]
|
|
if missing:
|
|
raise RegistryError(
|
|
f"{step_path} missing required fields: {', '.join(missing)}",
|
|
remediation=(
|
|
f"Add {', '.join(missing)} to {step_path}. Every onboarding step "
|
|
"must be self-describing for an operator who has no chat history."
|
|
),
|
|
source_path=source,
|
|
field_path=step_path,
|
|
)
|
|
state = _require_enum(
|
|
item.get("state", _DEFAULT_ONBOARDING_STATE),
|
|
allowed=ONBOARDING_STATES,
|
|
field_path=f"{step_path}.state",
|
|
source=source,
|
|
)
|
|
steps.append(
|
|
OnboardingStep(
|
|
id=str(item["id"]),
|
|
title=str(item["title"]),
|
|
description=str(item["description"]),
|
|
state=state,
|
|
required=bool(item.get("required", True)),
|
|
)
|
|
)
|
|
return tuple(steps)
|
|
|
|
|
|
def _parse_health(
|
|
raw: Any,
|
|
*,
|
|
project_path: str,
|
|
source: Path | None,
|
|
) -> ProjectHealth | None:
|
|
if raw is None:
|
|
return None
|
|
if not isinstance(raw, dict):
|
|
raise RegistryError(
|
|
f"{project_path}.last_seen_health must be an object when present",
|
|
remediation=(
|
|
f"Rewrite {project_path}.last_seen_health as an object with status "
|
|
f"(one of {', '.join(HEALTH_STATUSES)}), optional checked_at and detail, "
|
|
"or remove it. Never store endpoints or credentials here."
|
|
),
|
|
source_path=source,
|
|
field_path=f"{project_path}.last_seen_health",
|
|
)
|
|
status = _require_enum(
|
|
raw.get("status", "unknown"),
|
|
allowed=HEALTH_STATUSES,
|
|
field_path=f"{project_path}.last_seen_health.status",
|
|
source=source,
|
|
)
|
|
checked_at = raw.get("checked_at")
|
|
detail = raw.get("detail")
|
|
return ProjectHealth(
|
|
status=status,
|
|
checked_at=str(checked_at) if checked_at is not None else None,
|
|
detail=str(detail) if detail is not None else None,
|
|
)
|
|
|
|
|
|
def _parse_project(raw: Any, *, index: int, source: Path | None) -> ProjectRecord:
|
|
project_path = f"projects[{index}]"
|
|
if not isinstance(raw, dict):
|
|
raise RegistryError(
|
|
f"{project_path} must be an object",
|
|
remediation=f"Rewrite {project_path} as a JSON object describing one project.",
|
|
source_path=source,
|
|
field_path=project_path,
|
|
)
|
|
|
|
missing = [field for field in _REQUIRED_PROJECT_FIELDS if field not in raw]
|
|
if missing:
|
|
raise RegistryError(
|
|
f"{project_path} missing required fields: {', '.join(missing)}",
|
|
remediation=(
|
|
f"Add {', '.join(missing)} to {project_path}. See "
|
|
"docs/webui-project-registry-api.md for the field-by-field contract."
|
|
),
|
|
source_path=source,
|
|
field_path=project_path,
|
|
)
|
|
|
|
profiles = raw["profiles"]
|
|
if not isinstance(profiles, dict):
|
|
raise RegistryError(
|
|
f"{project_path}.profiles must be an object",
|
|
remediation=(
|
|
f"Rewrite {project_path}.profiles as an object mapping "
|
|
f"{', '.join(_REQUIRED_PROFILE_ROLES)} to MCP profile names."
|
|
),
|
|
source_path=source,
|
|
field_path=f"{project_path}.profiles",
|
|
)
|
|
for role in _REQUIRED_PROFILE_ROLES:
|
|
if role not in profiles or not profiles[role]:
|
|
raise RegistryError(
|
|
f"{project_path}.profiles.{role} is required",
|
|
remediation=(
|
|
f"Set {project_path}.profiles.{role} to the configured MCP profile "
|
|
"name for that role. Role separation is a workflow-safety invariant."
|
|
),
|
|
source_path=source,
|
|
field_path=f"{project_path}.profiles.{role}",
|
|
)
|
|
|
|
workflow_paths = raw["workflow_paths"]
|
|
if not isinstance(workflow_paths, dict) or not workflow_paths:
|
|
raise RegistryError(
|
|
f"{project_path}.workflow_paths must be a non-empty object",
|
|
remediation=(
|
|
f"Add at least a 'skill' entry to {project_path}.workflow_paths pointing "
|
|
"at the project's canonical workflow skill."
|
|
),
|
|
source_path=source,
|
|
field_path=f"{project_path}.workflow_paths",
|
|
)
|
|
|
|
schema_paths = raw.get("schema_paths") or {}
|
|
if not isinstance(schema_paths, dict):
|
|
raise RegistryError(
|
|
f"{project_path}.schema_paths must be an object when present",
|
|
remediation=(
|
|
f"Rewrite {project_path}.schema_paths as an object of label to repo path, "
|
|
"or remove it."
|
|
),
|
|
source_path=source,
|
|
field_path=f"{project_path}.schema_paths",
|
|
)
|
|
|
|
status = _require_enum(
|
|
raw.get("status", _DEFAULT_PROJECT_STATUS),
|
|
allowed=PROJECT_STATUSES,
|
|
field_path=f"{project_path}.status",
|
|
source=source,
|
|
)
|
|
remote_name = raw.get("remote_name")
|
|
|
|
return ProjectRecord(
|
|
id=str(raw["id"]),
|
|
repo_name=str(raw["repo_name"]),
|
|
gitea_owner=str(raw["gitea_owner"]),
|
|
remote_host=str(raw["remote_host"]),
|
|
default_branch=str(raw["default_branch"]),
|
|
local_checkout_path=str(raw["local_checkout_path"]),
|
|
profiles={role: str(profiles[role]) for role in _REQUIRED_PROFILE_ROLES},
|
|
workflow_paths={key: str(value) for key, value in workflow_paths.items()},
|
|
schema_paths={key: str(value) for key, value in schema_paths.items()},
|
|
onboarding_checklist=_parse_onboarding(
|
|
raw.get("onboarding_checklist"),
|
|
project_path=project_path,
|
|
source=source,
|
|
),
|
|
status=status,
|
|
remote_name=str(remote_name) if remote_name else None,
|
|
last_seen_health=_parse_health(
|
|
raw.get("last_seen_health"),
|
|
project_path=project_path,
|
|
source=source,
|
|
),
|
|
)
|
|
|
|
|
|
def load_registry(path: Path | None = None) -> ProjectRegistry:
|
|
"""Load the versioned project registry from disk.
|
|
|
|
Raises:
|
|
RegistryError: whenever the file is unreadable, is not valid JSON, or
|
|
fails schema validation. The error carries an operator remediation.
|
|
"""
|
|
source = (path or default_registry_path()).resolve()
|
|
try:
|
|
raw_text = source.read_text(encoding="utf-8")
|
|
except OSError as exc:
|
|
raise RegistryError(
|
|
f"registry file could not be read: {exc.strerror or exc}",
|
|
remediation=(
|
|
f"Create a readable registry at {source}, or point "
|
|
"WEBUI_PROJECT_REGISTRY at an existing file."
|
|
),
|
|
source_path=source,
|
|
) from exc
|
|
|
|
try:
|
|
payload = json.loads(raw_text)
|
|
except json.JSONDecodeError as exc:
|
|
raise RegistryError(
|
|
f"registry is not valid JSON: {exc.msg} (line {exc.lineno}, column {exc.colno})",
|
|
remediation=(
|
|
f"Fix the JSON syntax in {source} at line {exc.lineno}, column {exc.colno}."
|
|
),
|
|
source_path=source,
|
|
) from exc
|
|
|
|
if not isinstance(payload, dict):
|
|
raise RegistryError(
|
|
"registry root must be an object",
|
|
remediation=(
|
|
"Wrap the registry in a JSON object with 'version' and 'projects' keys."
|
|
),
|
|
source_path=source,
|
|
)
|
|
|
|
version = payload.get("version")
|
|
if version not in SUPPORTED_SCHEMA_VERSIONS:
|
|
supported = ", ".join(str(item) for item in SUPPORTED_SCHEMA_VERSIONS)
|
|
raise RegistryError(
|
|
f"unsupported registry version: {version!r}",
|
|
remediation=(
|
|
f"Set 'version' to one of {supported} (current schema is "
|
|
f"{CURRENT_SCHEMA_VERSION}). Migration notes live in "
|
|
"docs/webui-project-registry-api.md."
|
|
),
|
|
source_path=source,
|
|
field_path="version",
|
|
)
|
|
|
|
_reject_credential_keys(payload, source=source)
|
|
|
|
projects_raw = payload.get("projects")
|
|
if not isinstance(projects_raw, list) or not projects_raw:
|
|
raise RegistryError(
|
|
"projects must be a non-empty array",
|
|
remediation=(
|
|
"Add at least one project object to 'projects'. An empty console "
|
|
"registry fails closed rather than rendering a blank inventory."
|
|
),
|
|
source_path=source,
|
|
field_path="projects",
|
|
)
|
|
|
|
projects = tuple(
|
|
_parse_project(item, index=index, source=source)
|
|
for index, item in enumerate(projects_raw)
|
|
)
|
|
return ProjectRegistry(version=int(version), projects=projects, source_path=source)
|
|
|
|
|
|
def onboarding_summary(project: ProjectRecord) -> OnboardingSummary:
|
|
"""Aggregate a project's onboarding checklist state."""
|
|
steps = project.onboarding_checklist
|
|
counts = {state: 0 for state in ONBOARDING_STATES}
|
|
for step in steps:
|
|
counts[step.state] += 1
|
|
required_outstanding = sum(
|
|
1
|
|
for step in steps
|
|
if step.required and step.state in ("pending", "blocked")
|
|
)
|
|
return OnboardingSummary(
|
|
total=len(steps),
|
|
complete=counts["complete"],
|
|
pending=counts["pending"],
|
|
blocked=counts["blocked"],
|
|
not_applicable=counts["not_applicable"],
|
|
required_outstanding=required_outstanding,
|
|
onboarding_complete=required_outstanding == 0,
|
|
)
|
|
|
|
|
|
def project_to_dict(project: ProjectRecord) -> dict[str, Any]:
|
|
"""Serialize a project for JSON API responses and HTML views.
|
|
|
|
The HTML views render from this same DTO, so the console and the API can
|
|
never disagree about a project's status or onboarding progress.
|
|
"""
|
|
summary = onboarding_summary(project)
|
|
health = project.last_seen_health
|
|
return {
|
|
"id": project.id,
|
|
"repo_name": project.repo_name,
|
|
"gitea_owner": project.gitea_owner,
|
|
"repo_full_name": project.repo_full_name,
|
|
"remote_host": project.remote_host,
|
|
"remote_name": project.remote_name,
|
|
"default_branch": project.default_branch,
|
|
"local_checkout_path": project.local_checkout_path,
|
|
"status": project.status,
|
|
"profiles": dict(project.profiles),
|
|
"workflow_paths": dict(project.workflow_paths),
|
|
"schema_paths": dict(project.schema_paths),
|
|
"onboarding_checklist": [
|
|
{
|
|
"id": step.id,
|
|
"title": step.title,
|
|
"description": step.description,
|
|
"state": step.state,
|
|
"required": step.required,
|
|
}
|
|
for step in project.onboarding_checklist
|
|
],
|
|
"onboarding_summary": {
|
|
"total": summary.total,
|
|
"complete": summary.complete,
|
|
"pending": summary.pending,
|
|
"blocked": summary.blocked,
|
|
"not_applicable": summary.not_applicable,
|
|
"required_outstanding": summary.required_outstanding,
|
|
"onboarding_complete": summary.onboarding_complete,
|
|
},
|
|
"last_seen_health": (
|
|
None
|
|
if health is None
|
|
else {
|
|
"status": health.status,
|
|
"checked_at": health.checked_at,
|
|
"detail": health.detail,
|
|
}
|
|
),
|
|
}
|
|
|
|
|
|
def registry_to_dict(registry: ProjectRegistry) -> dict[str, Any]:
|
|
"""Serialize the whole registry, including API provenance (#632 section 6)."""
|
|
return {
|
|
"api_version": registry.api_version,
|
|
"schema_version": registry.schema_version,
|
|
# Retained for the unversioned MVP alias consumers (#427).
|
|
"version": registry.version,
|
|
"source_path": str(registry.source_path),
|
|
"source": {
|
|
"kind": "file",
|
|
"path": str(registry.source_path),
|
|
"inventory_complete": True,
|
|
},
|
|
"project_count": len(registry.projects),
|
|
"projects": [project_to_dict(project) for project in registry.projects],
|
|
}
|
|
|
|
|
|
def project_detail_to_dict(
|
|
registry: ProjectRegistry,
|
|
project: ProjectRecord,
|
|
) -> dict[str, Any]:
|
|
"""Serialize a single project for ``/api/v1/projects/{project_id}``."""
|
|
return {
|
|
"api_version": registry.api_version,
|
|
"schema_version": registry.schema_version,
|
|
"source": {
|
|
"kind": "file",
|
|
"path": str(registry.source_path),
|
|
"inventory_complete": True,
|
|
},
|
|
"project": project_to_dict(project),
|
|
}
|
|
|
|
|
|
def find_project(registry: ProjectRegistry, project_id: str) -> ProjectRecord | None:
|
|
for project in registry.projects:
|
|
if project.id == project_id:
|
|
return project
|
|
return None
|
|
|
|
|
|
def known_project_ids(registry: ProjectRegistry) -> list[str]:
|
|
return [project.id for project in registry.projects]
|