Files
govoplan/tools/inventory/platform-interface-inventory.py
zemion 2ffdb23f69
Dependency Audit / dependency-audit (push) Successful in 1m45s
Deployment Installer / deployment-installer (push) Successful in 6s
Security Audit / security-audit (push) Successful in 11m30s
feat(devkit): add resumable workspace automation and UI review tooling
Verified with the coordinated workspace changes by devkit full run
2026-09-08T225814-186389-0000-3e3ed7cd (all seven phases passed).
This shared UI pass does not mark the individual module reviews complete.
2026-09-09 02:03:17 +02:00

1389 lines
55 KiB
Python

#!/usr/bin/env python3
"""Extract a source-derived GovOPlaN UI, translation, module, and API inventory."""
from __future__ import annotations
import argparse
import ast
from collections import Counter
from dataclasses import asdict, is_dataclass
import importlib
import json
import os
from pathlib import Path
import re
import subprocess
import sys
from typing import Any
META_ROOT = Path(__file__).resolve().parents[2]
HTTP_METHODS = {"delete", "get", "head", "options", "patch", "post", "put"}
PATH_PARAMETER = re.compile(r"\$\{[^{}]*\}|\{[^{}]+\}")
ENDPOINT_SURFACE_CATEGORIES = {
"ui_reachable",
"intentionally_headless",
"public_integration",
"worker_internal",
"compatibility",
"missing_ui",
"removable",
}
DEFAULT_ENDPOINT_DECLARATIONS = (
META_ROOT / "tools" / "inventory" / "endpoint-surface-declarations.json"
)
DEFAULT_HIGH_RISK_HELP_BASELINE = (
META_ROOT / "tools" / "inventory" / "high-risk-help-baseline.json"
)
REQUIRED_LOCALES = ("de", "en")
REFERENCE_LOCALE = "de"
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--workspace-root",
type=Path,
help="Authoritative directory containing registered checkouts; never falls back to another workspace.",
)
parser.add_argument(
"--output-dir",
type=Path,
default=META_ROOT / "audit-reports" / "platform-inventory",
)
parser.add_argument(
"--strict",
action="store_true",
help="Fail on missing translations or incomplete endpoint-surface declarations.",
)
parser.add_argument(
"--strict-endpoints",
action="store_true",
help="Fail only on incomplete or stale endpoint-surface declarations.",
)
parser.add_argument(
"--strict-declarations",
action="store_true",
help=(
"Fail on duplicate stable IDs, WebUI surfaces absent from runtime "
"metadata, or stale runtime route declarations."
),
)
parser.add_argument(
"--runtime-snapshot",
type=Path,
help=(
"Compare a saved /api/v1/platform/interface-catalog response with "
"the static manifest inventory. Any installed module combination "
"is accepted; every module present in the snapshot must match."
),
)
parser.add_argument(
"--endpoint-declarations",
type=Path,
default=DEFAULT_ENDPOINT_DECLARATIONS,
help="Versioned endpoint-surface declaration registry.",
)
parser.add_argument(
"--high-risk-help-baseline",
type=Path,
default=DEFAULT_HIGH_RISK_HELP_BASELINE,
help="Versioned upper bound for high-risk controls without exact F1 help.",
)
args = parser.parse_args()
catalog = json.loads((META_ROOT / "repositories.json").read_text(encoding="utf-8"))
workspace_root = _resolve_workspace_root(catalog, args.workspace_root)
_validate_repository_roots(catalog, workspace_root)
webui = _extract_webui(workspace_root)
backend_endpoints = _extract_backend_endpoints(catalog, workspace_root)
manifests = _extract_manifests(catalog, workspace_root)
endpoint_declarations = _load_endpoint_declarations(
args.endpoint_declarations.resolve()
)
high_risk_help_baseline = _load_high_risk_help_baseline(
args.high_risk_help_baseline.resolve()
)
inventory = _assemble_inventory(
webui=webui,
backend_endpoints=backend_endpoints,
manifests=manifests,
endpoint_declarations=endpoint_declarations,
high_risk_help_baseline=high_risk_help_baseline,
runtime_snapshot=(
_load_runtime_snapshot(args.runtime_snapshot.resolve())
if args.runtime_snapshot is not None
else None
),
)
inventory["workspace_root"] = str(workspace_root)
inventory["workspace_selection"] = (
"explicit" if args.workspace_root is not None else "legacy-discovery"
)
output_dir = args.output_dir.resolve()
output_dir.mkdir(parents=True, exist_ok=True)
json_path = output_dir / "platform-interface-inventory.json"
markdown_path = output_dir / "platform-interface-inventory.md"
json_path.write_text(
json.dumps(inventory, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
markdown_path.write_text(_render_markdown(inventory), encoding="utf-8")
print(f"Platform inventory JSON: {json_path}")
print(f"Platform inventory summary: {markdown_path}")
if args.strict or args.strict_endpoints or args.strict_declarations:
failures = _strict_failures(
inventory,
check_translations=args.strict,
check_endpoints=args.strict or args.strict_endpoints,
check_declarations=args.strict or args.strict_declarations,
)
if failures:
print(
"Strict platform inventory failed: " + "; ".join(failures) + ".",
file=sys.stderr,
)
return 1
return 0
def _strict_failures(
inventory: dict[str, Any],
*,
check_translations: bool,
check_endpoints: bool,
check_declarations: bool = False,
) -> list[str]:
failures: list[str] = []
if (
check_translations
and inventory["translation_health"]["missing_catalog_entries"]
):
failures.append("used translation keys are missing from generated catalogs")
if check_endpoints and inventory["api"]["unclassified_endpoints"]:
failures.append(
f"{len(inventory['api']['unclassified_endpoints'])} backend "
"endpoints have no WebUI evidence or surface declaration"
)
if check_endpoints and inventory["api"]["stale_endpoint_declarations"]:
failures.append(
f"{len(inventory['api']['stale_endpoint_declarations'])} "
"endpoint declarations do not match a backend endpoint"
)
declaration_health = inventory.get("declaration_health", {})
if check_declarations and declaration_health.get("duplicate_ids"):
failures.append(
f"{len(declaration_health['duplicate_ids'])} platform interface "
"IDs are declared more than once"
)
if check_declarations and declaration_health.get("undeclared_source_surfaces"):
failures.append(
f"{len(declaration_health['undeclared_source_surfaces'])} public "
"WebUI surfaces have no runtime manifest declaration"
)
if check_declarations and declaration_health.get("stale_runtime_routes"):
failures.append(
f"{len(declaration_health['stale_runtime_routes'])} runtime route "
"declarations have no WebUI implementation"
)
help_health = inventory.get("help_health", {})
if check_declarations and help_health.get("invalid_risk_annotations"):
failures.append(
f"{len(help_health['invalid_risk_annotations'])} controls use an "
"unsupported contextual-help risk class"
)
if check_declarations and help_health.get("baseline_regression"):
failures.append(
f"{len(help_health['missing_exact_high_risk_help'])} high-risk "
"controls lack exact F1 help; baseline permits at most "
f"{help_health['baseline_maximum_missing']}"
)
if check_declarations and help_health.get("unresolved_exact_high_risk_help"):
failures.append(
f"{len(help_health['unresolved_exact_high_risk_help'])} high-risk "
"controls reference no manifest DocumentationTopic help context"
)
if check_declarations and help_health.get("high_risk_help_without_german"):
failures.append(
f"{len(help_health['high_risk_help_without_german'])} high-risk "
"controls resolve to documentation without complete German content"
)
runtime_comparison = inventory.get("runtime_comparison")
if (
check_declarations
and runtime_comparison is not None
and runtime_comparison.get("mismatches")
):
failures.append(
f"{len(runtime_comparison['mismatches'])} runtime catalog entries "
"do not match the release inventory"
)
return failures
def _resolve_workspace_root(
catalog: dict[str, Any], explicit_root: Path | None = None
) -> Path:
if explicit_root is not None:
root = explicit_root.expanduser().resolve()
if not root.is_dir():
raise ValueError("The explicit inventory workspace root must be an existing directory")
return root
# Compatibility for direct legacy callers only. Managed callers always
# supply their selected root; a partial checkout must not borrow sources.
sibling_root = META_ROOT.parent.resolve()
configured_root = Path(str(catalog["default_parent"])).expanduser().resolve()
repositories = catalog.get("repositories")
if not isinstance(repositories, list):
raise ValueError("repository catalog has no repositories array")
def source_count(root: Path) -> int:
return sum(
1
for item in repositories
if isinstance(item, dict)
and isinstance(item.get("path"), str)
and (root / item["path"] / "src").is_dir()
)
sibling_count = source_count(sibling_root)
configured_count = source_count(configured_root)
return sibling_root if sibling_count >= configured_count else configured_root
def _validate_repository_roots(catalog: dict[str, Any], workspace_root: Path) -> None:
repositories = catalog.get("repositories")
if not isinstance(repositories, list):
raise ValueError("repository catalog has no repositories array")
for repository in repositories:
if not isinstance(repository, dict) or not isinstance(repository.get("path"), str):
raise ValueError("Invalid inventory repository path")
relative = Path(repository["path"])
if not repository["path"] or relative.is_absolute() or ".." in relative.parts:
raise ValueError("Inventory repository paths must remain inside the selected workspace")
root = workspace_root / relative
# Missing optional checkouts are allowed; links to another checkout are
# not evidence for the selected workspace.
if not root.resolve().is_relative_to(workspace_root):
raise ValueError("Inventory repository path escapes the selected workspace")
for source in (root / "src", root / "webui/src"):
if not source.resolve().is_relative_to(workspace_root):
raise ValueError("Inventory source root escapes the selected workspace")
def _extract_webui(workspace_root: Path | None = None) -> dict[str, Any]:
helper = META_ROOT / "tools" / "inventory" / "extract-webui-structure.mjs"
argv = [os.environ.get("NODE", "node"), str(helper), str(META_ROOT)]
if workspace_root is not None:
argv.extend(["--workspace-root", str(workspace_root)])
completed = subprocess.run(
argv,
check=True,
capture_output=True,
text=True,
)
result = json.loads(completed.stdout)
if workspace_root is not None and (
not isinstance(result, dict)
or result.get("workspaceRoot") != str(workspace_root.resolve())
):
raise ValueError("WebUI collector did not confirm the selected inventory workspace")
return result
def _extract_backend_endpoints(
catalog: dict[str, Any],
workspace_root: Path,
) -> list[dict[str, Any]]:
endpoints: list[dict[str, Any]] = []
for repository in catalog["repositories"]:
repository_root = workspace_root / repository["path"]
source_root = repository_root / "src"
if not source_root.is_dir():
continue
for source_path in sorted(source_root.rglob("*.py")):
if not source_path.resolve().is_relative_to(workspace_root):
raise ValueError("Backend source path escapes the selected workspace")
try:
tree = ast.parse(
source_path.read_text(encoding="utf-8"),
filename=str(source_path),
)
except (OSError, SyntaxError, UnicodeDecodeError):
continue
prefixes = _router_prefixes(tree)
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
for decorator in node.decorator_list:
endpoint = _endpoint_from_decorator(
decorator,
prefixes=prefixes,
)
if endpoint is None:
continue
endpoints.append(
{
"repository": repository["name"],
"file": str(source_path.relative_to(repository_root)),
"line": node.lineno,
"handler": node.name,
**endpoint,
}
)
return sorted(
endpoints,
key=lambda item: (
item["repository"],
item["path"],
item["method"],
item["file"],
item["line"],
),
)
def _router_prefixes(tree: ast.AST) -> dict[str, str]:
prefixes: dict[str, str] = {}
for node in ast.walk(tree):
if not isinstance(node, (ast.Assign, ast.AnnAssign)):
continue
value = node.value
if not isinstance(value, ast.Call):
continue
function_name = _call_name(value.func)
if function_name not in {"APIRouter", "fastapi.APIRouter"}:
continue
prefix = ""
for keyword in value.keywords:
if keyword.arg == "prefix":
prefix = _static_string(keyword.value) or ""
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
for target in targets:
if isinstance(target, ast.Name):
prefixes[target.id] = prefix
return prefixes
def _endpoint_from_decorator(
decorator: ast.expr,
*,
prefixes: dict[str, str],
) -> dict[str, str] | None:
if not isinstance(decorator, ast.Call) or not isinstance(
decorator.func, ast.Attribute
):
return None
method = decorator.func.attr.lower()
if method not in HTTP_METHODS or not decorator.args:
return None
route = _static_string(decorator.args[0])
if route is None:
return None
owner = (
decorator.func.value.id if isinstance(decorator.func.value, ast.Name) else ""
)
prefix = prefixes.get(owner, "")
return {
"method": method.upper(),
"path": _join_route(prefix, route),
"router": owner,
}
def _extract_manifests(
catalog: dict[str, Any],
workspace_root: Path,
) -> list[dict[str, Any]]:
_validate_repository_roots(catalog, workspace_root)
source_roots = [
workspace_root / repository["path"] / "src"
for repository in catalog["repositories"]
if (workspace_root / repository["path"] / "src").is_dir()
]
core_root = next(
(workspace_root / repository["path"] / "src"
for repository in catalog["repositories"]
if repository.get("name") == "govoplan-core"),
workspace_root / "govoplan-core/src",
)
if not (core_root / "govoplan_core/core/platform_interfaces.py").is_file():
raise ValueError("Inventory requires Core interface sources in the selected workspace")
_assert_workspace_imports(workspace_root)
sys.path[:0] = [str(path) for path in source_roots]
from govoplan_core.core.platform_interfaces import ( # noqa: PLC0415
manifest_interface_catalog,
)
_assert_workspace_imports(workspace_root)
manifests: list[dict[str, Any]] = []
for repository in catalog["repositories"]:
source_root = workspace_root / repository["path"] / "src"
if not source_root.is_dir():
continue
for manifest_path in sorted(source_root.glob("*/backend/manifest.py")):
module_name = ".".join(
manifest_path.relative_to(source_root).with_suffix("").parts
)
loaded = importlib.import_module(module_name)
source = getattr(loaded, "__file__", None)
if not isinstance(source, str) or Path(source).resolve() != manifest_path.resolve():
raise ValueError("Manifest import did not resolve to its selected workspace source")
_assert_workspace_imports(workspace_root)
manifest = loaded.get_manifest()
_assert_workspace_imports(workspace_root)
frontend = manifest.frontend
manifests.append(
{
"repository": repository["name"],
"id": manifest.id,
"name": manifest.name,
"version": manifest.version,
"dependencies": list(manifest.dependencies),
"optional_dependencies": list(manifest.optional_dependencies),
"required_capabilities": list(manifest.required_capabilities),
"provided_interfaces": [
{"name": item.name, "version": item.version}
for item in manifest.provides_interfaces
],
"runtime_capabilities": sorted(manifest.capability_factories),
"permissions": [
{
"scope": permission.scope,
"label": permission.label,
"level": permission.level,
}
for permission in manifest.permissions
],
"documentation": [
{
"id": topic.id,
"help_contexts": sorted(
{
str(context)
for context in topic.metadata.get(
"help_contexts", ()
)
if isinstance(context, str) and context.strip()
}
),
"german_complete": (
isinstance(topic.translations.get("de"), dict)
and all(
isinstance(topic.translations["de"].get(field), str)
and topic.translations["de"][field].strip()
for field in ("title", "summary", "body")
)
),
}
for topic in manifest.documentation
],
"architecture": (
manifest.architecture.to_dict()
if manifest.architecture is not None
else None
),
"information_governance": (
manifest.information_governance.to_dict()
),
"interface_catalog": manifest_interface_catalog(manifest),
"frontend": (
{
"package": frontend.package_name,
"routes": [
_plain_value(route) for route in frontend.routes
],
"public_routes": [
_plain_value(route)
for route in frontend.public_routes
],
"nav_items": [
_plain_value(item) for item in frontend.nav_items
],
"settings_routes": [
_plain_value(route)
for route in frontend.settings_routes
],
"view_surfaces": [
_plain_value(surface)
for surface in frontend.view_surfaces
],
}
if frontend is not None
else None
),
}
)
return sorted(manifests, key=lambda item: item["id"])
def _assert_workspace_imports(workspace_root: Path) -> None:
# An editable installation or cached import must not stand in for a missing
# optional checkout. Direct callers with another workspace use a fresh
# process instead of replacing already-loaded application packages.
for name, module in list(sys.modules.items()):
# The Meta tools may audit a separate checkout; they are not module
# contributions and must not be confused with application packages.
package = name.partition(".")[0]
if not package.startswith("govoplan_") or package in {
"govoplan_devkit", "govoplan_release"
}:
continue
filename = getattr(module, "__file__", None)
locations = list(getattr(module, "__path__", ()))
if isinstance(filename, str):
locations.append(filename)
if any(not Path(location).resolve().is_relative_to(workspace_root) for location in locations):
raise ValueError("A GovOPlaN import originates outside the selected inventory workspace")
def _assemble_inventory(
*,
webui: dict[str, Any],
backend_endpoints: list[dict[str, Any]],
manifests: list[dict[str, Any]],
endpoint_declarations: dict[tuple[str, str, str], dict[str, Any]],
high_risk_help_baseline: dict[str, Any] | None = None,
runtime_snapshot: dict[str, Any] | None = None,
) -> dict[str, Any]:
frontend_refs = webui["frontendApiReferences"]
frontend_paths = {
canonical_api_path(reference["path"])
for reference in frontend_refs
if canonical_api_path(reference["path"])
}
endpoint_keys = {endpoint_key(endpoint) for endpoint in backend_endpoints}
classified_endpoints: list[dict[str, Any]] = []
for endpoint in backend_endpoints:
key = endpoint_key(endpoint)
canonical_path = key[2]
static_webui_reference = canonical_path in frontend_paths
declaration = endpoint_declarations.get(key)
surface = (
{
"category": "ui_reachable",
"rationale": "A canonical API path reference exists in WebUI source.",
"source": "static_webui_scan",
}
if static_webui_reference
else (
{**declaration, "source": "declaration"}
if declaration is not None
else None
)
)
classified_endpoints.append(
{
**endpoint,
"canonical_path": canonical_path,
"static_webui_reference": static_webui_reference,
"surface": surface,
}
)
unreferenced = [
endpoint
for endpoint in classified_endpoints
if not endpoint["static_webui_reference"]
]
unclassified = [
endpoint for endpoint in unreferenced if endpoint["surface"] is None
]
stale_declarations = [
declaration
for key, declaration in endpoint_declarations.items()
if key not in endpoint_keys
]
classification_counts = Counter(
endpoint["surface"]["category"]
for endpoint in classified_endpoints
if endpoint["surface"] is not None
)
usages = {item["key"] for item in webui["translationUsages"]}
catalogs = webui["translationCatalog"]
catalog_keys = {locale: set(entries) for locale, entries in catalogs.items()}
expected_locales = sorted(set(catalog_keys) | set(REQUIRED_LOCALES))
missing_catalog_entries = [
{
"key": key,
"missing_locales": [
locale
for locale in expected_locales
if key not in catalog_keys.get(locale, set())
],
}
for key in sorted(usages)
if any(key not in catalog_keys.get(locale, set()) for locale in expected_locales)
]
fields = webui["fields"]
actions = webui.get("actions", [])
help_candidates = [field for field in fields if field["helpCandidate"]]
dynamic_help = [field for field in fields if field.get("helpDynamic")]
controls = [*fields, *actions]
high_risk_controls = [item for item in controls if item.get("helpRisk")]
missing_exact_high_risk_help = [
item for item in controls if item.get("highRiskHelpMissing")
]
invalid_risk_annotations = [
item
for item in controls
if item.get("helpRiskSource") == "invalid_explicit"
]
documentation_contexts = {
context: {
"module_id": manifest["id"],
"topic_id": topic["id"],
"german_complete": topic["german_complete"],
}
for manifest in manifests
for topic in manifest.get("documentation", [])
for context in topic.get("help_contexts", [])
}
unresolved_exact_high_risk_help = [
item
for item in high_risk_controls
if item.get("helpExact")
and not item.get("helpContextDynamic")
and item.get("helpContextId") not in documentation_contexts
]
high_risk_help_without_german = [
item
for item in high_risk_controls
if item.get("helpContextId") in documentation_contexts
and not documentation_contexts[item["helpContextId"]]["german_complete"]
]
baseline_maximum_missing = (
high_risk_help_baseline["maximum_missing_exact_help"]
if high_risk_help_baseline is not None
else None
)
governance_adoption = Counter(
dimension["adoption"]
for manifest in manifests
for dimension in manifest["information_governance"]["dimensions"].values()
)
source_declarations = _source_interface_declarations(webui, manifests)
declaration_health = _declaration_health(source_declarations, manifests)
runtime_comparison = (
_compare_runtime_snapshot(runtime_snapshot, manifests)
if runtime_snapshot is not None
else None
)
return {
"schema_version": 2,
"scope": {
"source": "local GovOPlaN repository catalog",
"limitations": [
"Static extraction cannot resolve runtime-computed labels, routes, or API paths.",
"A backend endpoint without a static WebUI reference may intentionally serve public clients, workers, connectors, or external integrations.",
"A field marked as a help candidate may receive contextual help from a surrounding dynamic component.",
"Low-level field and action IDs use line-independent source anchors unless an explicit interfaceId, DOM id, or name is declared.",
],
},
"modules": manifests,
"interface_declarations": source_declarations,
"declaration_health": declaration_health,
"help_health": {
"supported_risk_classes": sorted(
{
str(item["helpRisk"])
for item in high_risk_controls
if item.get("helpRisk")
}
),
"high_risk_controls": high_risk_controls,
"missing_exact_high_risk_help": missing_exact_high_risk_help,
"invalid_risk_annotations": invalid_risk_annotations,
"unresolved_exact_high_risk_help": unresolved_exact_high_risk_help,
"high_risk_help_without_german": high_risk_help_without_german,
"dynamic_owner_context_controls": [
item
for item in high_risk_controls
if item.get("helpContextDynamic")
],
"baseline_maximum_missing": baseline_maximum_missing,
"baseline_regression": (
baseline_maximum_missing is not None
and len(missing_exact_high_risk_help) > baseline_maximum_missing
),
},
"runtime_comparison": runtime_comparison,
"ui": {
"fields": fields,
"actions": actions,
"labels": webui["labels"],
"visible_text": webui["visibleText"],
"routes": webui["routes"],
"navigation": webui["navigation"],
"ui_capabilities": webui["uiCapabilities"],
"help_candidates": help_candidates,
"dynamic_help": dynamic_help,
"contributions": webui.get("contributions", []),
},
"translations": {
"catalog": catalogs,
"usages": webui["translationUsages"],
"dynamic_usages": webui["dynamicTranslationUsages"],
},
"translation_health": {
"locales": expected_locales,
"reference_locale": REFERENCE_LOCALE,
"reference_locale_entries": len(catalog_keys.get(REFERENCE_LOCALE, set())),
"reference_locale_complete": not any(
REFERENCE_LOCALE in item["missing_locales"]
for item in missing_catalog_entries
),
"used_keys": len(usages),
"missing_catalog_entries": missing_catalog_entries,
},
"information_governance_health": {
"dimensions": len(manifests) * 4,
"adoption_counts": dict(sorted(governance_adoption.items())),
"modules": [
{
"module_id": manifest["id"],
"dimensions": manifest["information_governance"]["dimensions"],
}
for manifest in manifests
],
},
"api": {
"backend_endpoints": classified_endpoints,
"frontend_references": frontend_refs,
"unreferenced_by_static_webui_scan": unreferenced,
"unclassified_endpoints": unclassified,
"stale_endpoint_declarations": stale_declarations,
"classification_counts": dict(sorted(classification_counts.items())),
},
"summary": {
"modules": len(manifests),
"ui_fields": len(fields),
"ui_fields_with_static_help": len(fields) - len(help_candidates),
"ui_fields_with_resolvable_f1_context": len(fields),
"help_review_candidates": len(help_candidates),
"dynamic_help_references": len(dynamic_help),
"ui_actions": len(actions),
"high_risk_controls": len(high_risk_controls),
"high_risk_controls_with_exact_help": (
len(high_risk_controls) - len(missing_exact_high_risk_help)
),
"high_risk_controls_missing_exact_help": len(
missing_exact_high_risk_help
),
"invalid_help_risk_annotations": len(invalid_risk_annotations),
"unresolved_exact_high_risk_help": len(
unresolved_exact_high_risk_help
),
"high_risk_help_without_german": len(high_risk_help_without_german),
"interface_declarations": len(source_declarations),
"duplicate_interface_ids": len(declaration_health["duplicate_ids"]),
"undeclared_source_surfaces": len(
declaration_health["undeclared_source_surfaces"]
),
"stale_runtime_routes": len(
declaration_health["stale_runtime_routes"]
),
"label_attributes": len(webui["labels"]),
"visible_text_nodes": len(webui["visibleText"]),
"frontend_routes": len(webui["routes"]),
"backend_endpoints": len(backend_endpoints),
"frontend_api_references": len(frontend_refs),
"backend_endpoints_without_static_webui_reference": len(unreferenced),
"unclassified_backend_endpoints": len(unclassified),
"stale_endpoint_declarations": len(stale_declarations),
"information_governance_dimensions": len(manifests) * 4,
"information_governance_enforced": governance_adoption["enforced"],
"information_governance_partial": governance_adoption["partial"],
"information_governance_contract_only": governance_adoption[
"contract_only"
],
},
}
def _source_interface_declarations(
webui: dict[str, Any],
manifests: list[dict[str, Any]],
) -> list[dict[str, Any]]:
module_by_repository = {
str(manifest["repository"]): str(manifest["id"])
for manifest in manifests
}
declarations: list[dict[str, Any]] = []
def module_id(repository: str) -> str:
if repository in module_by_repository:
return module_by_repository[repository]
if repository == "govoplan-core":
return "core"
return repository.removeprefix("govoplan-").replace("-", "_")
def source_evidence(item: dict[str, Any]) -> dict[str, Any]:
return {
key: item[key]
for key in ("repository", "file", "line", "column")
if key in item
}
for kind, items in (
("field", webui["fields"]),
("action", webui.get("actions", [])),
):
for item in items:
repository = str(item["repository"])
owner = module_id(repository)
raw_id = str(item["id"])
stable_id = (
f"{owner}.{raw_id[len(repository) + 1:]}"
if raw_id.startswith(f"{repository}.")
else _namespaced_interface_id(owner, kind, raw_id)
)
declarations.append(
{
"key": f"{kind}:{stable_id}",
"id": stable_id,
"module_id": owner,
"kind": kind,
"origin": "webui_source",
"id_source": item.get("idSource", "source_anchor"),
"explicit_id": item.get("explicitId"),
"context": item.get("context"),
**source_evidence(item),
}
)
if kind == "field" and item.get("helpId"):
help_raw_id = str(item["helpId"])
help_id = (
f"{owner}.{help_raw_id[len(repository) + 1:]}"
if help_raw_id.startswith(f"{repository}.")
else _namespaced_interface_id(owner, "help", help_raw_id)
)
declarations.append(
{
"key": f"help:{help_id}",
"id": help_id,
"module_id": owner,
"kind": "help",
"origin": "webui_source",
"field_id": stable_id,
"dynamic": bool(item.get("helpDynamic")),
**source_evidence(item),
}
)
for item in webui.get("contributions", []):
repository = str(item["repository"])
owner = module_id(repository)
kind = str(item["kind"])
raw_id = str(item["id"])
path = item.get("path")
if kind == "frontend_route" and isinstance(path, str):
stable_id = f"{owner}.route.{_surface_slug(path)}"
elif kind == "public_route" and isinstance(path, str):
stable_id = f"{owner}.public.{_surface_slug(path)}"
elif kind == "navigation" and isinstance(path, str):
stable_id = f"{owner}.nav.{_surface_slug(path)}"
else:
stable_id = _namespaced_interface_id(owner, kind, raw_id)
declarations.append(
{
"key": f"{kind}:{stable_id}",
"id": stable_id,
"module_id": owner,
"kind": kind,
"origin": "webui_contribution",
"declared_value": raw_id,
**({"path": path} if isinstance(path, str) else {}),
**source_evidence(item),
}
)
translation_entries: dict[tuple[str, str], dict[str, Any]] = {}
for locale, entries in webui["translationCatalog"].items():
for key, item in entries.items():
repository = str(item["repository"])
owner = module_id(repository)
declaration_key = (owner, str(key))
declaration = translation_entries.setdefault(
declaration_key,
{
"key": f"translation:{key}",
"id": key,
"module_id": owner,
"kind": "translation",
"origin": "translation_catalog",
"locales": [],
**source_evidence(item),
},
)
declaration["locales"].append(locale)
declarations.extend(translation_entries.values())
return sorted(
declarations,
key=lambda item: (
item["module_id"],
item["kind"],
item["id"],
item.get("file", ""),
item.get("line", 0),
),
)
def _declaration_health(
source_declarations: list[dict[str, Any]],
manifests: list[dict[str, Any]],
) -> dict[str, Any]:
grouped: dict[tuple[str, str, str], list[dict[str, Any]]] = {}
for declaration in source_declarations:
key = (
str(declaration["module_id"]),
str(declaration["kind"]),
str(declaration["id"]),
)
grouped.setdefault(key, []).append(declaration)
duplicate_ids = [
{
"module_id": key[0],
"kind": key[1],
"id": key[2],
"evidence": values,
}
for key, values in sorted(grouped.items())
if len(values) > 1
]
comparable_kinds = {
"frontend_route",
"navigation",
"public_route",
"view_surface",
}
source_surfaces = {
(str(item["module_id"]), str(item["kind"]), str(item["id"])): item
for item in source_declarations
if item["origin"] == "webui_contribution"
and item["kind"] in comparable_kinds
}
runtime_surfaces: dict[tuple[str, str, str], dict[str, Any]] = {}
for manifest in manifests:
catalog = manifest["interface_catalog"]
for declaration in catalog["declarations"]:
if declaration["kind"] not in comparable_kinds:
continue
key = (
str(manifest["id"]),
str(declaration["kind"]),
str(declaration["id"]),
)
runtime_surfaces[key] = {
"repository": manifest["repository"],
**declaration,
}
surface_id = declaration.get("metadata", {}).get("surface_id")
if (
declaration["kind"]
in {"frontend_route", "navigation", "settings_route"}
and isinstance(surface_id, str)
and surface_id
):
runtime_surfaces[
(str(manifest["id"]), "view_surface", surface_id)
] = {
"repository": manifest["repository"],
"key": f"view_surface:{surface_id}",
"id": surface_id,
"module_id": manifest["id"],
"kind": "view_surface",
"metadata": {
"derived_from": declaration["kind"],
"path": declaration.get("path"),
},
}
undeclared_source_surfaces = [
source_surfaces[key]
for key in sorted(source_surfaces.keys() - runtime_surfaces.keys())
]
route_kinds = {"frontend_route", "public_route"}
stale_runtime_routes = [
runtime_surfaces[key]
for key in sorted(runtime_surfaces.keys() - source_surfaces.keys())
if key[1] in route_kinds
]
return {
"duplicate_ids": duplicate_ids,
"undeclared_source_surfaces": undeclared_source_surfaces,
"stale_runtime_routes": stale_runtime_routes,
"source_declaration_count": len(source_declarations),
"runtime_declaration_count": sum(
len(manifest["interface_catalog"]["declarations"])
for manifest in manifests
),
}
def _compare_runtime_snapshot(
snapshot: dict[str, Any],
manifests: list[dict[str, Any]],
) -> dict[str, Any]:
static_by_module = {
str(manifest["id"]): manifest["interface_catalog"]
for manifest in manifests
}
modules = snapshot.get("modules")
if not isinstance(modules, list):
raise ValueError("Runtime interface snapshot must contain a modules list.")
mismatches: list[dict[str, Any]] = []
seen: set[str] = set()
matched: list[str] = []
for item in modules:
if not isinstance(item, dict) or not isinstance(item.get("module_id"), str):
raise ValueError("Runtime interface snapshot has an invalid module entry.")
module_id = item["module_id"]
if module_id in seen:
mismatches.append({"module_id": module_id, "reason": "duplicate_module"})
continue
seen.add(module_id)
expected = static_by_module.get(module_id)
if expected is None:
mismatches.append({"module_id": module_id, "reason": "unknown_module"})
continue
for field in ("contract_version", "module_version", "digest"):
if item.get(field) != expected.get(field):
mismatches.append(
{
"module_id": module_id,
"reason": f"{field}_mismatch",
"expected": expected.get(field),
"actual": item.get(field),
}
)
if not any(
mismatch["module_id"] == module_id for mismatch in mismatches
):
matched.append(module_id)
return {
"contract_version": snapshot.get("contract_version"),
"matched_modules": sorted(matched),
"mismatches": mismatches,
}
def _load_runtime_snapshot(path: Path) -> dict[str, Any]:
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError as exc:
raise ValueError(f"Runtime interface snapshot does not exist: {path}") from exc
except json.JSONDecodeError as exc:
raise ValueError(f"Runtime interface snapshot is invalid JSON: {exc}") from exc
if not isinstance(payload, dict):
raise ValueError("Runtime interface snapshot must be a JSON object.")
return payload
def _namespaced_interface_id(module_id: str, kind: str, value: str) -> str:
if value.startswith(f"{module_id}."):
return value
return f"{module_id}.{kind}.{_surface_slug(value)}"
def _surface_slug(value: str) -> str:
normalized = re.sub(r"[^a-z0-9]+", ".", value.strip().lower()).strip(".")
return normalized or "root"
def _render_markdown(inventory: dict[str, Any]) -> str:
summary = inventory["summary"]
missing = inventory["translation_health"]["missing_catalog_entries"]
help_by_repository = Counter(
item["repository"] for item in inventory["ui"]["help_candidates"]
)
endpoint_by_repository = Counter(
item["repository"]
for item in inventory["api"]["unreferenced_by_static_webui_scan"]
)
classification_counts = inventory["api"]["classification_counts"]
high_risk_by_repository = Counter(
item["repository"]
for item in inventory["help_health"]["missing_exact_high_risk_help"]
)
lines = [
"# GovOPlaN Platform Interface Inventory",
"",
"Generated from runtime module manifests plus TypeScript and Python ASTs.",
"Counts are source evidence, not a claim that every surface is enabled or reachable.",
"",
"## Summary",
"",
f"- Modules: {summary['modules']}",
f"- UI fields: {summary['ui_fields']}",
f"- UI actions: {summary['ui_actions']}",
f"- Fields with statically associated help: {summary['ui_fields_with_static_help']}",
f"- Fields with a resolvable F1 context: {summary['ui_fields_with_resolvable_f1_context']}",
f"- Fields with dynamic help references: {summary['dynamic_help_references']}",
f"- Help review candidates: {summary['help_review_candidates']}",
f"- High-risk controls: {summary['high_risk_controls']}",
f"- High-risk controls with exact F1 help: {summary['high_risk_controls_with_exact_help']}",
f"- High-risk controls missing exact F1 help: {summary['high_risk_controls_missing_exact_help']}",
f"- Invalid help-risk annotations: {summary['invalid_help_risk_annotations']}",
f"- High-risk exact contexts missing a manifest topic: {summary['unresolved_exact_high_risk_help']}",
f"- High-risk contexts without complete German topic content: {summary['high_risk_help_without_german']}",
f"- Stable interface declarations: {summary['interface_declarations']}",
f"- Duplicate interface IDs: {summary['duplicate_interface_ids']}",
f"- WebUI surfaces missing runtime declarations: {summary['undeclared_source_surfaces']}",
f"- Runtime routes missing WebUI implementations: {summary['stale_runtime_routes']}",
f"- Label attributes: {summary['label_attributes']}",
f"- Frontend routes: {summary['frontend_routes']}",
f"- Backend endpoints: {summary['backend_endpoints']}",
f"- Frontend API references: {summary['frontend_api_references']}",
(
"- Backend endpoints without a static WebUI reference: "
f"{summary['backend_endpoints_without_static_webui_reference']}"
),
f"- Unclassified backend endpoints: {summary['unclassified_backend_endpoints']}",
f"- Stale endpoint declarations: {summary['stale_endpoint_declarations']}",
f"- Reference locale: `{inventory['translation_health']['reference_locale']}`",
f"- Reference locale complete: `{str(inventory['translation_health']['reference_locale_complete']).lower()}`",
f"- Used translation keys missing from a required locale catalog: {len(missing)}",
f"- Information-governance dimensions enforced: {summary['information_governance_enforced']}",
f"- Information-governance dimensions partial: {summary['information_governance_partial']}",
f"- Information-governance dimensions contract-only: {summary['information_governance_contract_only']}",
"",
"## Help Review Candidates",
"",
"| Repository | Fields |",
"| --- | ---: |",
]
lines.extend(
f"| `{repository}` | {count} |"
for repository, count in sorted(help_by_repository.items())
)
lines.extend(
[
"",
"## High-risk Contextual-help Debt",
"",
"Inferred or explicitly classified high-risk controls require an exact",
"F1 context. `data-help-risk-reviewed=\"standard\"` records a reviewed",
"false positive. The versioned baseline makes this queue non-regressing.",
"",
"| Repository | Missing exact contexts |",
"| --- | ---: |",
]
)
lines.extend(
f"| `{repository}` | {count} |"
for repository, count in sorted(high_risk_by_repository.items())
)
lines.extend(
[
"",
"## Endpoints Without Static WebUI References",
"",
"These are review candidates. Public APIs, worker callbacks, connector",
"endpoints, health checks, and dynamically assembled paths are legitimate",
"reasons for appearing here.",
"",
"| Repository | Endpoints |",
"| --- | ---: |",
]
)
lines.extend(
f"| `{repository}` | {count} |"
for repository, count in sorted(endpoint_by_repository.items())
)
lines.extend(
[
"",
"## Endpoint Surface Classifications",
"",
"| Classification | Endpoints |",
"| --- | ---: |",
]
)
lines.extend(
f"| `{category}` | {count} |"
for category, count in sorted(classification_counts.items())
)
lines.extend(
[
"",
"## Declaration Reconciliation",
"",
"Routes, navigation, View surfaces, fields, actions, help references,",
"translations, settings, widgets, search objects, and backend",
"capabilities use normalized stable IDs. CI rejects duplicate IDs,",
"WebUI public surfaces absent from runtime metadata, and runtime routes",
"without a WebUI implementation.",
"",
"",
"## Interpretation",
"",
"Use the JSON artifact for exact file and line evidence. Missing help is",
"a triage list, not an automatic defect. Endpoint coverage requires an",
"owner classification before enforcement. Runtime-computed structures",
"need explicit manifest or typed PlatformWebModule metadata to become",
"canonically visible. The generated files are release evidence, not an",
"editable source of platform behavior.",
"",
]
)
return "\n".join(lines)
def canonical_api_path(value: str) -> str:
path = value.split("?", 1)[0].strip()
if "/api/" in path:
path = path[path.index("/api/") :]
path = re.sub(r"^/api/v\d+", "", path)
path = re.sub(r"(?<=[^/])\$\{[^{}]*\}$", "", path)
path = PATH_PARAMETER.sub("{}", path)
path = re.sub(r"/+", "/", path)
return path.rstrip("/") or "/"
def endpoint_key(endpoint: dict[str, Any]) -> tuple[str, str, str]:
return (
str(endpoint["repository"]),
str(endpoint["method"]).upper(),
canonical_api_path(str(endpoint["path"])),
)
def _load_high_risk_help_baseline(path: Path) -> dict[str, Any]:
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError as exc:
raise ValueError(
f"High-risk contextual-help baseline does not exist: {path}"
) from exc
except json.JSONDecodeError as exc:
raise ValueError(
f"High-risk contextual-help baseline is invalid JSON: {exc}"
) from exc
if not isinstance(payload, dict) or payload.get("schema_version") != 1:
raise ValueError(
"High-risk contextual-help baseline must use schema_version 1."
)
maximum = payload.get("maximum_missing_exact_help")
if not isinstance(maximum, int) or isinstance(maximum, bool) or maximum < 0:
raise ValueError(
"High-risk contextual-help baseline maximum must be a non-negative integer."
)
return payload
def _load_endpoint_declarations(
path: Path,
) -> dict[tuple[str, str, str], dict[str, Any]]:
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError as exc:
raise ValueError(
f"Endpoint declaration registry does not exist: {path}"
) from exc
except json.JSONDecodeError as exc:
raise ValueError(
f"Endpoint declaration registry is invalid JSON: {exc}"
) from exc
if not isinstance(payload, dict) or payload.get("schema_version") != 1:
raise ValueError("Endpoint declaration registry must use schema_version 1.")
entries = payload.get("endpoints")
if not isinstance(entries, list):
raise ValueError(
"Endpoint declaration registry must contain an endpoints list."
)
declarations: dict[tuple[str, str, str], dict[str, Any]] = {}
for index, entry in enumerate(entries):
if not isinstance(entry, dict):
raise ValueError(f"Endpoint declaration {index} must be an object.")
repository = entry.get("repository")
method = entry.get("method")
raw_path = entry.get("path")
category = entry.get("category")
rationale = entry.get("rationale")
if not isinstance(repository, str) or not repository.strip():
raise ValueError(f"Endpoint declaration {index} has no repository.")
if not isinstance(method, str) or method.lower() not in HTTP_METHODS:
raise ValueError(f"Endpoint declaration {index} has an invalid method.")
if not isinstance(raw_path, str) or not raw_path.startswith("/"):
raise ValueError(f"Endpoint declaration {index} has an invalid path.")
canonical_path = canonical_api_path(raw_path)
if raw_path != canonical_path:
raise ValueError(
f"Endpoint declaration {index} path must be canonical: {canonical_path}"
)
if category not in ENDPOINT_SURFACE_CATEGORIES:
raise ValueError(f"Endpoint declaration {index} has an invalid category.")
if not isinstance(rationale, str) or not rationale.strip():
raise ValueError(f"Endpoint declaration {index} has no rationale.")
tracking_issue = entry.get("tracking_issue")
if category == "missing_ui" and (
not isinstance(tracking_issue, str) or not tracking_issue.strip()
):
raise ValueError(
f"Endpoint declaration {index} requires a tracking_issue for missing_ui."
)
key = (repository, method.upper(), canonical_path)
if key in declarations:
raise ValueError(f"Duplicate endpoint declaration: {key!r}.")
declarations[key] = {
"repository": repository,
"method": method.upper(),
"path": canonical_path,
"category": category,
"rationale": rationale.strip(),
**(
{"tracking_issue": tracking_issue.strip()}
if isinstance(tracking_issue, str) and tracking_issue.strip()
else {}
),
}
return declarations
def _join_route(prefix: str, route: str) -> str:
return f"/{prefix.strip('/')}/{route.strip('/')}".replace("//", "/")
def _static_string(node: ast.AST) -> str | None:
if isinstance(node, ast.Constant) and isinstance(node.value, str):
return node.value
if isinstance(node, ast.JoinedStr):
pieces: list[str] = []
for value in node.values:
if isinstance(value, ast.Constant) and isinstance(value.value, str):
pieces.append(value.value)
elif isinstance(value, ast.FormattedValue):
pieces.append("${}")
else:
return None
return "".join(pieces)
return None
def _call_name(node: ast.AST) -> str:
if isinstance(node, ast.Name):
return node.id
if isinstance(node, ast.Attribute):
parent = _call_name(node.value)
return f"{parent}.{node.attr}" if parent else node.attr
return ""
def _plain_value(value: Any) -> Any:
if is_dataclass(value):
return {key: _plain_value(item) for key, item in asdict(value).items()}
if isinstance(value, tuple):
return [_plain_value(item) for item in value]
if isinstance(value, dict):
return {str(key): _plain_value(item) for key, item in value.items()}
return value
if __name__ == "__main__":
raise SystemExit(main())