aureon-os / scripts /validation /generate_system_integration_map.py
GaryLeckey's picture
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 19)
da0ce7b verified
Raw History Blame Contribute Delete
15.7 kB
"""Generate Aureon's repo-wide system integration map."""
from __future__ import annotations
import json
import subprocess
from collections import Counter, defaultdict
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[2]
SNAPSHOT_DATE = "2026-07-13"
DOCS_REPO_SITEMAP = REPO_ROOT / "docs" / "repo_sitemap.json"
DOCS_ACCESS_MAP = REPO_ROOT / "docs" / "end_user_access_map.json"
DOCS_CAPABILITY_REGISTRY = REPO_ROOT / "docs" / "capability_registry.json"
DOCS_NAVIGATION_INDEX = REPO_ROOT / "docs" / "repo_navigation_index.json"
DOCS_SAAS_MANIFEST = REPO_ROOT / "docs" / "saas_integration_manifest.json"
DOCS_HARDENING_MANIFEST = REPO_ROOT / "docs" / "supabase_hardening_manifest.json"
DOCS_MANIFEST = REPO_ROOT / "docs" / "system_integration_map.json"
PUBLIC_MANIFEST = REPO_ROOT / "frontend" / "public" / "aureon_system_integration_map.json"
PUBLIC_ARTIFACT_PREFIXES = ("frontend/public/", "public/")
VALIDATION_PREFIXES = ("tests/", "scripts/validation/", "VERIFICATION AND VALIDATION/")
ENTRYPOINT_KINDS = {
"markdown",
"json",
"config",
"python",
"typescript",
"typescript-react",
"javascript",
"javascript-react",
"windows-command",
"powershell",
"shell",
"sql",
}
SYSTEM_OVERRIDES = {
".github/": {
"integration_role": "repository automation and CI metadata",
"safety_gate": "keep repository automation scoped and review dependency/security alerts",
"access_mode": "GitHub repository metadata",
},
"api/": {
"integration_role": "hosted API route surface",
"safety_gate": "confirm auth, CORS, and payload validation before public exposure",
"access_mode": "hosted/serverless API",
},
"aureon/": {
"integration_role": "core local runtime and domain automation package",
"safety_gate": "live mutation routes require operator review, credentials, and dry-run/live gates",
"access_mode": "local runtime and internal package",
},
"data/": {
"integration_role": "evidence, research, grant, and dataset store",
"safety_gate": "redact private evidence before public disclosure",
"access_mode": "public-safe evidence or controlled local files",
},
"docs/": {
"integration_role": "primary reading, diligence, runbook, and generated-map system",
"safety_gate": "keep claims evidence-backed and generated maps current",
"access_mode": "public documentation",
},
"frontend/": {
"integration_role": "React/Vite user console and public manifest surface",
"safety_gate": "build and render-check public manifests before publishing",
"access_mode": "browser application",
},
"Kings_Accounting_Suite/": {
"integration_role": "accounting, filing-support, and statutory-pack tooling",
"safety_gate": "filing and payment stay manual unless separately authorized",
"access_mode": "local operator workflow",
},
"scripts/": {
"integration_role": "operator diagnostics, reports, launch helpers, and validators",
"safety_gate": "run focused validation before relying on generated artifacts",
"access_mode": "local maintainer tooling",
},
"supabase/": {
"integration_role": "SaaS backend, Edge Functions, auth settings, and migrations",
"safety_gate": "resolve public Edge Function hardening blockers before hosted production",
"access_mode": "hosted Supabase backend",
},
"tests/": {
"integration_role": "regression, contract, and behavior verification",
"safety_gate": "checks must cover the claimed capability or integration surface",
"access_mode": "local or CI validation",
},
}
def git_ls_files() -> list[str]:
result = subprocess.run(
["git", "ls-files", "-z"],
cwd=REPO_ROOT,
check=True,
capture_output=True,
)
return [item.decode("utf-8", errors="replace") for item in result.stdout.split(b"\0") if item]
def load_json(path: Path) -> dict:
with path.open("r", encoding="utf-8") as handle:
return json.load(handle)
def normalize_path(path: str) -> str:
return path if path.endswith("/") else f"{path}/"
def path_matches(path: str, prefix: str) -> bool:
normalized = prefix.rstrip("/")
return path == normalized or path.startswith(f"{normalized}/")
def capability_lookup(access_map: dict) -> dict[str, set[str]]:
lookup: dict[str, set[str]] = defaultdict(set)
for capability in access_map.get("capabilities", []):
capability_id = str(capability.get("id", ""))
for field in ("primary_docs", "related_systems", "runtime_or_api_surface"):
for target in capability.get(field, []):
if not isinstance(target, str) or target.startswith(("http://", "https://", "#", "/")):
continue
lookup[target].add(capability_id)
return lookup
def top_level_system_for_path(path: str, system_paths: set[str]) -> str | None:
normalized = path.strip().lstrip("/")
if not normalized or normalized.startswith(("http://", "https://", "#")):
return None
if normalized.endswith("/") and normalized in system_paths:
return normalized
if "/" not in normalized:
return None
top_level = normalized.split("/", 1)[0] + "/"
return top_level if top_level in system_paths else None
def capability_system_lookup(capability_registry: dict, access_map: dict, system_paths: set[str]) -> dict[str, set[str]]:
lookup: dict[str, set[str]] = defaultdict(set)
routes_by_id = {
route["id"]: route
for route in access_map.get("capabilities", [])
if isinstance(route, dict) and route.get("id")
}
for capability in capability_registry.get("capabilities", []):
if not isinstance(capability, dict) or not capability.get("id"):
continue
capability_id = str(capability["id"])
candidate_paths: list[str] = []
candidate_paths.extend(capability.get("system_paths", []))
candidate_paths.extend(capability.get("resolved_paths", []))
candidate_paths.extend(capability.get("generated_refs", []))
candidate_paths.extend(capability.get("public_artifacts", []))
for route_id in capability.get("access_route_ids", []):
route = routes_by_id.get(route_id)
if not route:
continue
candidate_paths.extend(route.get("primary_docs", []))
candidate_paths.extend(route.get("related_systems", []))
for candidate in candidate_paths:
if not isinstance(candidate, str):
continue
system_path = top_level_system_for_path(candidate, system_paths)
if system_path:
lookup[system_path].add(capability_id)
return lookup
def access_routes_for_system(system_path: str, lookup: dict[str, set[str]], entries: list[dict]) -> list[str]:
ids: set[str] = set()
for prefix, capability_ids in lookup.items():
if path_matches(prefix, system_path) or path_matches(system_path, prefix):
ids.update(capability_ids)
for entry in entries:
if path_matches(entry.get("path", ""), system_path):
ids.update(entry.get("capability_ids", []))
return sorted(ids)
def pick_entrypoints(system_path: str, entries: list[dict]) -> list[str]:
candidates: list[str] = []
for entry in entries:
path = entry.get("path", "")
if not path_matches(path, system_path):
continue
if entry.get("kind") in ENTRYPOINT_KINDS:
name = Path(path).name.lower()
if name in {"readme.md", "index.md"} or path.count("/") <= 2:
candidates.append(path)
return candidates[:12]
def pick_public_artifacts(system_path: str, entries: list[dict]) -> list[str]:
artifacts = [
entry.get("path", "")
for entry in entries
if path_matches(entry.get("path", ""), system_path)
and entry.get("path", "").startswith(PUBLIC_ARTIFACT_PREFIXES)
and entry.get("kind") in {"json", "markdown", "html", "image"}
]
return artifacts[:15]
def pick_validation_refs(system_path: str, entries: list[dict], capability_ids: list[str]) -> list[str]:
refs: list[str] = []
for entry in entries:
path = entry.get("path", "")
if not path.startswith(VALIDATION_PREFIXES):
continue
entry_capabilities = set(entry.get("capability_ids", []))
if path_matches(path, system_path) or entry_capabilities.intersection(capability_ids):
refs.append(path)
return refs[:12]
def readiness_for(system_path: str, category: str) -> str:
if system_path in {"frontend/", "docs/", "data/", "scripts/", "tests/"}:
return "mapped_for_end_user_navigation"
if system_path in {"supabase/", "api/", "server/", "functions/", "netlify/", "deploy/", "production/"}:
return "integration_ready_with_production_gates"
if category in {"archive", "provenance"}:
return "preserved_for_audit_not_current_product_surface"
if category in {"runtime", "back office"}:
return "operator_controlled_before_saas_exposure"
return "mapped_with_review_gate"
def build_manifest() -> dict:
repo_sitemap = load_json(DOCS_REPO_SITEMAP)
access_map = load_json(DOCS_ACCESS_MAP)
capability_registry = load_json(DOCS_CAPABILITY_REGISTRY)
navigation_index = load_json(DOCS_NAVIGATION_INDEX)
saas_manifest = load_json(DOCS_SAAS_MANIFEST)
hardening_manifest = load_json(DOCS_HARDENING_MANIFEST)
tracked_paths = git_ls_files()
entries = navigation_index.get("entries", [])
access_lookup = capability_lookup(access_map)
system_paths = {normalize_path(system["path"]) for system in repo_sitemap.get("top_level_systems", [])}
implementation_lookup = capability_system_lookup(capability_registry, access_map, system_paths)
capability_counts = Counter()
access_route_counts = Counter()
systems = []
for system in repo_sitemap.get("top_level_systems", []):
system_path = normalize_path(system["path"])
category = system.get("category", "uncategorized")
override = SYSTEM_OVERRIDES.get(system_path, {})
access_route_ids = access_routes_for_system(system_path, access_lookup, entries)
capability_ids = sorted(implementation_lookup.get(system_path, set()))
for capability_id in capability_ids:
capability_counts[capability_id] += 1
for route_id in access_route_ids:
access_route_counts[route_id] += 1
systems.append(
{
"path": system_path,
"category": category,
"tracked_files": system.get("files", 0),
"role": system.get("role", ""),
"integration_role": override.get("integration_role", system.get("role", "")),
"access_mode": override.get("access_mode", "repo path and generated navigation"),
"readiness_status": readiness_for(system_path, category),
"capability_ids": capability_ids,
"access_route_ids": access_route_ids,
"entrypoints": pick_entrypoints(system_path, entries),
"public_artifacts": pick_public_artifacts(system_path, entries),
"validation_refs": pick_validation_refs(system_path, entries, [*capability_ids, *access_route_ids]),
"safety_gate": override.get("safety_gate", "review before exposing as a hosted or mutable surface"),
}
)
registry_capability_ids = sorted(
str(capability.get("id", ""))
for capability in capability_registry.get("capabilities", [])
if isinstance(capability, dict) and capability.get("id")
)
unmapped_capability_ids = sorted(set(registry_capability_ids) - set(capability_counts))
return {
"name": "Aureon System Integration Map",
"schema_version": 1,
"snapshot_date": SNAPSHOT_DATE,
"generated_by": "scripts/validation/generate_system_integration_map.py",
"docs_mirror": "docs/system_integration_map.json",
"frontend_public_mirror": "frontend/public/aureon_system_integration_map.json",
"source_documents": [
"docs/repo_sitemap.json",
"docs/end_user_access_map.json",
"docs/capability_registry.json",
"docs/repo_navigation_index.json",
"docs/saas_integration_manifest.json",
"docs/supabase_hardening_manifest.json",
],
"public_contract": {
"contains_file_contents": False,
"contains_secrets": False,
"contains_private_runtime_state": False,
"contains_customer_data": False,
},
"summary": {
"tracked_file_count": len(tracked_paths),
"system_count": len(systems),
"capability_count": len(registry_capability_ids),
"mapped_capability_count": len(capability_counts),
"unmapped_capability_count": len(unmapped_capability_ids),
"capability_system_binding_count": sum(capability_counts.values()),
"access_route_count": len(access_map.get("capabilities", [])),
"mapped_access_route_count": len(access_route_counts),
"access_route_system_binding_count": sum(access_route_counts.values()),
"saas_deployment_surface_count": len(saas_manifest.get("deployment_surfaces", [])),
"supabase_public_blocker_count": hardening_manifest.get("summary", {}).get("production_blocker_count", 0),
},
"systems": systems,
"capability_system_counts": dict(sorted(capability_counts.items())),
"access_route_system_counts": dict(sorted(access_route_counts.items())),
"unmapped_capability_ids": unmapped_capability_ids,
"integration_sequence": [
"start with README.md, docs/REPO_SITEMAP.md, and docs/END_USER_ACCESS_MAP.md",
"use docs/system_integration_map.json to bind systems to capabilities and gates",
"use docs/repo_navigation_index.json for file-level lookup",
"use docs/saas_integration_manifest.json for deployment/env/auth surfaces",
"use docs/supabase_hardening_manifest.json before hosted Supabase production",
"run scripts/validation/validate_repo_navigation_contract.py before publishing navigation changes",
],
"frontend_navigation_tab": "#repo-map",
}
def write_json(path: Path, data: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data, ensure_ascii=False, separators=(",", ":")) + "\n", encoding="utf-8")
def main() -> int:
manifest = build_manifest()
write_json(DOCS_MANIFEST, manifest)
write_json(PUBLIC_MANIFEST, manifest)
print(f"Wrote {DOCS_MANIFEST.relative_to(REPO_ROOT)}")
print(f"Wrote {PUBLIC_MANIFEST.relative_to(REPO_ROOT)}")
print(
"Mapped "
f"{manifest['summary']['system_count']} systems across "
f"{manifest['summary']['mapped_capability_count']} capability bindings"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())