93b8174abd
Close CUST-WP-0064 after the 2026-08-24 unassisted fire ingested clay-borg, close CUST-WP-0065 now that all 120 active repos project a classification, and close ADHOC-2026-08-25. Mark CUST-WP-0067 T02/T10 done (reverse relays already gone; work-record recovery lives on 0068). Park the later no-checkout SBOM regression as CUST-IN-0015. Teach the classification gate to use this host's checkout path.
237 lines
8.5 KiB
Python
237 lines
8.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Report active-fleet Repo Classification convergence from State Hub.
|
|
|
|
The repository file remains authoritative. This command compares the live
|
|
State Hub projection with each active record's registered local checkout and
|
|
separates missing owner decisions from projection failures.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
import yaml
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
from tools.validate_repo_classification import load_allowed, validate # noqa: E402
|
|
|
|
DEFAULT_API_BASE = "http://127.0.0.1:8000"
|
|
UrlOpen = Callable[..., Any]
|
|
|
|
|
|
class FleetReadError(RuntimeError):
|
|
"""State Hub did not return a usable repository projection."""
|
|
|
|
|
|
def fetch_repositories(
|
|
api_base: str,
|
|
*,
|
|
timeout: float = 15.0,
|
|
opener: UrlOpen = urllib.request.urlopen,
|
|
) -> list[dict[str, Any]]:
|
|
url = f"{api_base.rstrip('/')}/repos/?limit=500"
|
|
request = urllib.request.Request(url, headers={"Accept": "application/json"})
|
|
try:
|
|
with opener(request, timeout=timeout) as response:
|
|
payload = json.load(response)
|
|
except (OSError, urllib.error.URLError, json.JSONDecodeError) as exc:
|
|
raise FleetReadError(f"cannot read State Hub repositories: {exc}") from exc
|
|
if not isinstance(payload, list) or any(not isinstance(item, dict) for item in payload):
|
|
raise FleetReadError("State Hub /repos response is not a list of objects")
|
|
return payload
|
|
|
|
|
|
def _checkout_candidates(record: dict[str, Any], *, hostname: str) -> list[Path]:
|
|
"""Prefer this host's recorded checkout over a `local_path` from another machine."""
|
|
seen: set[str] = set()
|
|
candidates: list[Path] = []
|
|
|
|
def add(raw: object) -> None:
|
|
if not isinstance(raw, str):
|
|
return
|
|
text = raw.strip()
|
|
if not text or text in seen:
|
|
return
|
|
seen.add(text)
|
|
candidates.append(Path(text))
|
|
|
|
host_paths = record.get("host_paths")
|
|
if isinstance(host_paths, dict):
|
|
short = hostname.split(".")[0]
|
|
add(host_paths.get(hostname))
|
|
add(host_paths.get(short))
|
|
for value in host_paths.values():
|
|
if isinstance(value, str) and Path(value).is_dir():
|
|
add(value)
|
|
add(record.get("local_path"))
|
|
return candidates
|
|
|
|
|
|
def _source_path(record: dict[str, Any], *, hostname: str) -> Path | None:
|
|
preferred: Path | None = None
|
|
for checkout in _checkout_candidates(record, hostname=hostname):
|
|
path = checkout / ".repo-classification.yaml"
|
|
if preferred is None:
|
|
preferred = path
|
|
if path.is_file():
|
|
return path
|
|
return preferred
|
|
|
|
|
|
def analyze_repositories(
|
|
records: list[dict[str, Any]],
|
|
*,
|
|
validate_sources: bool = True,
|
|
hostname: str | None = None,
|
|
) -> dict[str, Any]:
|
|
active = [record for record in records if record.get("status") == "active"]
|
|
classified = [record for record in active if record.get("category") is not None]
|
|
null_records = [record for record in active if record.get("category") is None]
|
|
missing_source: list[str] = []
|
|
present_unprojected: list[str] = []
|
|
projected_without_source: list[str] = []
|
|
invalid_source: dict[str, list[str]] = {}
|
|
source_warnings: dict[str, list[str]] = {}
|
|
duplicate_slugs: list[str] = []
|
|
|
|
slug_counts: dict[str, int] = {}
|
|
allowed = load_allowed() if validate_sources else None
|
|
current_host = hostname or socket.gethostname()
|
|
for record in active:
|
|
slug = str(record.get("slug") or "<missing-slug>")
|
|
slug_counts[slug] = slug_counts.get(slug, 0) + 1
|
|
path = _source_path(record, hostname=current_host)
|
|
source_present = path is not None and path.is_file()
|
|
|
|
if record.get("category") is None:
|
|
(present_unprojected if source_present else missing_source).append(slug)
|
|
elif not source_present:
|
|
projected_without_source.append(slug)
|
|
|
|
if validate_sources and source_present and path is not None:
|
|
try:
|
|
document = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
errors, warnings = validate(document, allowed)
|
|
except (OSError, yaml.YAMLError) as exc:
|
|
errors, warnings = [f"cannot read/parse source: {exc}"], []
|
|
if errors:
|
|
invalid_source[slug] = errors
|
|
if warnings:
|
|
source_warnings[slug] = warnings
|
|
|
|
duplicate_slugs = sorted(slug for slug, count in slug_counts.items() if count > 1)
|
|
missing_source.sort()
|
|
present_unprojected.sort()
|
|
projected_without_source.sort()
|
|
|
|
blocking = bool(
|
|
missing_source
|
|
or present_unprojected
|
|
or projected_without_source
|
|
or invalid_source
|
|
or duplicate_slugs
|
|
)
|
|
return {
|
|
"registered": len(records),
|
|
"active": len(active),
|
|
"active_classified": len(classified),
|
|
"active_null_category": len(null_records),
|
|
"missing_source": missing_source,
|
|
"present_unprojected": present_unprojected,
|
|
"projected_without_source": projected_without_source,
|
|
"invalid_source": dict(sorted(invalid_source.items())),
|
|
"source_warnings": dict(sorted(source_warnings.items())),
|
|
"source_warning_count": sum(len(items) for items in source_warnings.values()),
|
|
"duplicate_active_slugs": duplicate_slugs,
|
|
"converged": not blocking,
|
|
}
|
|
|
|
|
|
def render_text(report: dict[str, Any], *, show_warnings: bool = False) -> str:
|
|
lines = [
|
|
f"registered: {report['registered']}",
|
|
f"active: {report['active']}",
|
|
f"active classified: {report['active_classified']}",
|
|
f"active null category: {report['active_null_category']}",
|
|
f"missing source file: {len(report['missing_source'])}",
|
|
f"source present but unprojected: {len(report['present_unprojected'])}",
|
|
f"projected without source file: {len(report['projected_without_source'])}",
|
|
f"invalid source file: {len(report['invalid_source'])}",
|
|
f"source validation warnings: {report['source_warning_count']}",
|
|
f"duplicate active slug: {len(report['duplicate_active_slugs'])}",
|
|
f"converged: {'yes' if report['converged'] else 'no'}",
|
|
]
|
|
for key in (
|
|
"missing_source",
|
|
"present_unprojected",
|
|
"projected_without_source",
|
|
"duplicate_active_slugs",
|
|
):
|
|
if report[key]:
|
|
lines.append(f"{key}: {', '.join(report[key])}")
|
|
for slug, errors in report["invalid_source"].items():
|
|
lines.append(f"invalid_source[{slug}]: {'; '.join(errors)}")
|
|
if show_warnings:
|
|
for slug, warnings in report["source_warnings"].items():
|
|
lines.append(f"source_warnings[{slug}]: {'; '.join(warnings)}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--api-base",
|
|
default=os.environ.get("STATE_HUB_API_BASE", DEFAULT_API_BASE),
|
|
help="State Hub REST base URL (default: %(default)s)",
|
|
)
|
|
parser.add_argument("--timeout", type=float, default=15.0)
|
|
parser.add_argument("--json", action="store_true", help="emit machine-readable JSON")
|
|
parser.add_argument(
|
|
"--show-warnings",
|
|
action="store_true",
|
|
help="include individual non-blocking source validation warnings",
|
|
)
|
|
parser.add_argument(
|
|
"--skip-source-validation",
|
|
action="store_true",
|
|
help="only compare file presence and projection state",
|
|
)
|
|
parser.add_argument(
|
|
"--require-converged",
|
|
action="store_true",
|
|
help="exit 1 while any active classification gap remains",
|
|
)
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
try:
|
|
records = fetch_repositories(args.api_base, timeout=args.timeout)
|
|
report = analyze_repositories(
|
|
records,
|
|
validate_sources=not args.skip_source_validation,
|
|
)
|
|
except FleetReadError as exc:
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return 2
|
|
|
|
if args.json:
|
|
print(json.dumps(report, indent=2, sort_keys=True))
|
|
else:
|
|
print(render_text(report, show_warnings=args.show_warnings))
|
|
return 1 if args.require_converged and not report["converged"] else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|