feat(governance): add product manifest standard readiness

apps/api/src/services/product_awoooi_manifest_standard.py

@@ -0,0 +1,157 @@
+"""P0-002 product.awoooi.yaml manifest standard readiness."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from src.services.snapshot_paths import default_operations_dir, resolve_repo_root
+
+_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
+_SNAPSHOT_FILE = "product-awoooi-manifest-standard.snapshot.json"
+_SCHEMA_VERSION = "product_awoooi_manifest_standard_readiness_v1"
+_MANIFEST_SCHEMA_VERSION = "product_awoooi_manifest_v1"
+
+
+def load_latest_product_awoooi_manifest_standard(
+    operations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load and validate the committed P0-002 product manifest standard."""
+    directory = operations_dir or _DEFAULT_OPERATIONS_DIR
+    path = directory / _SNAPSHOT_FILE
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path}: expected JSON object")
+    repo_root = resolve_repo_root(path)
+    _enrich_manifest_readiness(payload, repo_root)
+    _require_schema(payload, str(path))
+    _require_operation_boundaries(payload, str(path))
+    _require_rollup_consistency(payload, str(path))
+    return payload
+
+
+def _enrich_manifest_readiness(payload: dict[str, Any], repo_root: Path) -> None:
+    manifest_meta = _dict(payload.get("manifest"))
+    manifest_path = repo_root / str(manifest_meta.get("path") or "")
+    schema_path = repo_root / str(manifest_meta.get("schema_ref") or "")
+    manifest = _load_yaml_object(manifest_path)
+
+    manifest_meta["present"] = manifest_path.is_file()
+    manifest_meta["schema_present"] = schema_path.is_file()
+    manifest_meta["schema_version"] = str(manifest.get("schema_version") or "")
+    manifest_meta["product_id"] = str(_dict(manifest.get("product")).get("id") or "")
+    manifest_meta["source_control_authority"] = str(
+        _dict(manifest.get("source_control")).get("authority") or ""
+    )
+    manifest_meta["github_status"] = str(
+        _dict(manifest.get("source_control")).get("github_status") or ""
+    )
+
+    sections = _strings(payload.get("required_sections"))
+    missing_sections = [section for section in sections if section not in manifest]
+    refs = _strings(payload.get("required_contract_refs"))
+    missing_refs = [ref for ref in refs if not (repo_root / ref).is_file()]
+
+    rollups = _dict(payload.get("rollups"))
+    rollups["required_section_count"] = len(sections)
+    rollups["present_required_section_count"] = len(sections) - len(missing_sections)
+    rollups["missing_required_section_count"] = len(missing_sections)
+    rollups["missing_required_sections"] = missing_sections
+    rollups["required_contract_ref_count"] = len(refs)
+    rollups["present_contract_ref_count"] = len(refs) - len(missing_refs)
+    rollups["missing_contract_ref_count"] = len(missing_refs)
+    rollups["missing_contract_refs"] = missing_refs
+    rollups["hard_blocker_count"] = len(_list(payload.get("blockers")))
+    rollups["next_action_count"] = len(_list(payload.get("next_actions")))
+    ready_count = rollups["present_required_section_count"] + rollups[
+        "present_contract_ref_count"
+    ]
+    required_count = rollups["required_section_count"] + rollups[
+        "required_contract_ref_count"
+    ]
+    rollups["source_readiness_percent"] = round(
+        ready_count / max(required_count, 1) * 100
+    )
+    payload["status"] = (
+        "ready_for_product_manifest_adoption"
+        if not missing_sections
+        and not missing_refs
+        and manifest_meta["present"]
+        and manifest_meta["schema_present"]
+        and manifest_meta["schema_version"] == _MANIFEST_SCHEMA_VERSION
+        and manifest_meta["source_control_authority"] == "gitea"
+        and manifest_meta["github_status"] == "stopped_retired_do_not_use"
+        else "blocked_product_manifest_source_missing"
+    )
+
+
+def _load_yaml_object(path: Path) -> dict[str, Any]:
+    if not path.is_file():
+        return {}
+    with path.open(encoding="utf-8") as handle:
+        loaded = yaml.safe_load(handle)
+    if not isinstance(loaded, dict):
+        raise ValueError(f"{path}: expected YAML object")
+    return loaded
+
+
+def _require_schema(payload: dict[str, Any], label: str) -> None:
+    actual = payload.get("schema_version")
+    if actual != _SCHEMA_VERSION:
+        raise ValueError(
+            f"{label}: expected schema_version={_SCHEMA_VERSION}, got {actual!r}"
+        )
+
+
+def _require_operation_boundaries(payload: dict[str, Any], label: str) -> None:
+    boundaries = _dict(payload.get("operation_boundaries"))
+    if boundaries.get("read_only_api_allowed") is not True:
+        raise ValueError(f"{label}: read_only_api_allowed must be true")
+    if boundaries.get("manifest_write_allowed") is not True:
+        raise ValueError(f"{label}: manifest_write_allowed must be true for P0-002 source")
+    blocked_flags = {
+        "remaining_product_repo_write_allowed",
+        "repo_creation_allowed",
+        "refs_sync_allowed",
+        "workflow_trigger_allowed",
+        "github_api_allowed",
+        "host_or_k8s_write_allowed",
+        "secret_read_allowed",
+        "raw_session_or_sqlite_read_allowed",
+    }
+    allowed = sorted(flag for flag in blocked_flags if boundaries.get(flag) is not False)
+    if allowed:
+        raise ValueError(f"{label}: operation boundaries must remain false: {allowed}")
+
+
+def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
+    rollups = _dict(payload.get("rollups"))
+    sections = _strings(payload.get("required_sections"))
+    refs = _strings(payload.get("required_contract_refs"))
+    blockers = _list(payload.get("blockers"))
+    next_actions = _list(payload.get("next_actions"))
+    if rollups.get("required_section_count") != len(sections):
+        raise ValueError(f"{label}: required_section_count mismatch")
+    if rollups.get("required_contract_ref_count") != len(refs):
+        raise ValueError(f"{label}: required_contract_ref_count mismatch")
+    if rollups.get("hard_blocker_count") != len(blockers):
+        raise ValueError(f"{label}: hard_blocker_count mismatch")
+    if rollups.get("next_action_count") != len(next_actions):
+        raise ValueError(f"{label}: next_action_count mismatch")
+
+
+def _dict(value: Any) -> dict[str, Any]:
+    return value if isinstance(value, dict) else {}
+
+
+def _list(value: Any) -> list[Any]:
+    return value if isinstance(value, list) else []
+
+
+def _strings(value: Any) -> list[str]:
+    return [str(item) for item in _list(value) if str(item)]