feat(work-items): surface product governance matrix

apps/api/src/api/v1/agents.py

@@ -386,12 +386,12 @@ from src.services.dependency_upgrade_approval_package_template import (
 from src.services.docker_build_surface_inventory import (
     load_latest_docker_build_surface_inventory,
 )
-from src.services.gitea_authenticated_inventory_payload_validation import (
-    validate_gitea_authenticated_inventory_payload,
-)
 from src.services.gitea_actions_secret_variable_name_inventory import (
     load_latest_gitea_actions_secret_variable_name_inventory,
 )
+from src.services.gitea_authenticated_inventory_payload_validation import (
+    validate_gitea_authenticated_inventory_payload,
+)
 from src.services.gitea_branch_protection_required_checks_readback import (
     load_latest_gitea_branch_protection_required_checks_readback,
 )
@@ -461,6 +461,9 @@ from src.services.product_awoooi_manifest_standard import (
 from src.services.product_code_review_gate import (
     load_latest_product_code_review_gate,
 )
+from src.services.product_governance_matrix_readback import (
+    load_latest_product_governance_matrix_readback,
+)
 from src.services.public_redaction import redact_public_lan_topology
 from src.services.reboot_auto_recovery_drill_preflight import (
     load_latest_reboot_auto_recovery_drill_preflight,
@@ -1875,6 +1878,38 @@ async def get_gitea_private_inventory_p0_scorecard() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/product-governance-matrix-readback",
+    response_model=dict[str, Any],
+    summary="取得全產品 product governance matrix 讀回",
+    description=(
+        "整合已提交的 Gitea private inventory 與 dev/prod repo truth，"
+        "輸出每個產品的 canonical repo、main/dev branch、visibility、"
+        "owner readiness 與 backup evidence ref。此端點只讀 snapshot；"
+        "不呼叫 GitHub、不建立 repo、不改 visibility、不同步 refs、"
+        "不觸發 workflow、不讀 secret、不讀 raw session/SQLite。"
+    ),
+)
+async def get_product_governance_matrix_readback() -> dict[str, Any]:
+    """回傳全產品 Gitea source-truth governance matrix。"""
+    try:
+        payload = await asyncio.to_thread(
+            load_latest_product_governance_matrix_readback
+        )
+        return redact_public_lan_topology(payload)
+    except FileNotFoundError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except (json.JSONDecodeError, ValueError) as exc:
+        logger.error("product_governance_matrix_readback_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Product governance matrix readback 快照無效",
+        ) from exc
+
+
 @router.post(
     "/gitea-private-inventory-p0-scorecard/validate-authenticated-inventory-payload",
     response_model=dict[str, Any],

apps/api/src/services/product_governance_matrix_readback.py

@@ -0,0 +1,317 @@
+"""Product governance matrix readback.
+
+This service promotes committed Gitea source-truth snapshots into a product
+governance matrix. It is read-only: no GitHub calls, no Gitea writes, no secret
+reads, no workflow dispatch, and no runtime host action.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+from zoneinfo import ZoneInfo
+
+from src.services.gitea_private_inventory_p0_scorecard import (
+    load_latest_gitea_private_inventory_p0_scorecard,
+)
+from src.services.snapshot_paths import default_operations_dir
+
+SCHEMA_VERSION = "product_governance_matrix_readback_v1"
+TZ_TAIPEI = ZoneInfo("Asia/Taipei")
+_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
+_PRIVATE_INVENTORY_FILE = "awoooi-gitea-private-inventory-p0-scorecard.snapshot.json"
+_DEV_PROD_REPO_READBACK_FILE = "gitea-all-product-dev-prod-repo-readback.snapshot.json"
+
+
+def load_latest_product_governance_matrix_readback(
+    operations_dir: Path | None = None,
+    *,
+    repo_bundle_backup_readback: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Load the committed product source-truth matrix."""
+    directory = operations_dir or _DEFAULT_OPERATIONS_DIR
+    private_inventory_path = directory / _PRIVATE_INVENTORY_FILE
+    dev_prod_path = directory / _DEV_PROD_REPO_READBACK_FILE
+    private_inventory_source = _load_json_object(private_inventory_path)
+    dev_prod_source = _load_json_object(dev_prod_path)
+    private_inventory = load_latest_gitea_private_inventory_p0_scorecard(directory)
+    return build_product_governance_matrix_readback(
+        private_inventory_source=private_inventory_source,
+        dev_prod_repo_readback=dev_prod_source,
+        private_inventory_readback=private_inventory,
+        repo_bundle_backup_readback=repo_bundle_backup_readback,
+        source_refs={
+            "private_inventory": f"docs/operations/{private_inventory_path.name}",
+            "dev_prod_repo_readback": f"docs/operations/{dev_prod_path.name}",
+            "repo_bundle_backup_readback": (
+                "GET /api/v1/agents/gitea-repo-bundle-backup-readback"
+            ),
+        },
+    )
+
+
+def build_product_governance_matrix_readback(
+    *,
+    private_inventory_source: dict[str, Any],
+    dev_prod_repo_readback: dict[str, Any],
+    private_inventory_readback: dict[str, Any],
+    repo_bundle_backup_readback: dict[str, Any] | None = None,
+    generated_at: str | None = None,
+    source_refs: dict[str, str] | None = None,
+) -> dict[str, Any]:
+    """Build a matrix from source-control truth and optional backup evidence."""
+    refs = source_refs or {}
+    private_rollups = _dict(private_inventory_readback.get("rollups"))
+    dev_prod_summary = _dict(dev_prod_repo_readback.get("summary"))
+    product_rows = _list(private_inventory_source.get("product_rows"))
+    bundle_rows = _bundle_repo_index(repo_bundle_backup_readback)
+    bundle_summary = _dict((repo_bundle_backup_readback or {}).get("summary"))
+    bundle_ready = (repo_bundle_backup_readback or {}).get("ready") is True
+
+    products = [
+        _product_row(
+            row,
+            bundle_rows=bundle_rows,
+            bundle_ready=bundle_ready,
+            bundle_ref=refs.get(
+                "repo_bundle_backup_readback",
+                "GET /api/v1/agents/gitea-repo-bundle-backup-readback",
+            ),
+        )
+        for row in product_rows
+    ]
+    ready_count = sum(1 for row in products if row["source_control_ready"])
+    blocker_ids = sorted(
+        {
+            blocker
+            for row in products
+            for blocker in _strings(row.get("blockers"))
+        }
+    )
+    all_rows_ready = len(products) > 0 and ready_count == len(products)
+    status_value = (
+        "product_governance_matrix_ready"
+        if all_rows_ready and not blocker_ids
+        else "product_governance_matrix_action_required"
+    )
+
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "generated_at": generated_at
+        or datetime.now(TZ_TAIPEI).isoformat(timespec="seconds"),
+        "status": status_value,
+        "scope": "product_governance_matrix",
+        "source_control_authority": "gitea",
+        "source_refs": {
+            "private_inventory": refs.get(
+                "private_inventory",
+                f"docs/operations/{_PRIVATE_INVENTORY_FILE}",
+            ),
+            "dev_prod_repo_readback": refs.get(
+                "dev_prod_repo_readback",
+                f"docs/operations/{_DEV_PROD_REPO_READBACK_FILE}",
+            ),
+            "repo_bundle_backup_readback": refs.get(
+                "repo_bundle_backup_readback",
+                "GET /api/v1/agents/gitea-repo-bundle-backup-readback",
+            ),
+        },
+        "summary": {
+            "product_count": len(products),
+            "ready_product_count": ready_count,
+            "blocked_product_count": len(products) - ready_count,
+            "expected_product_count": _int(
+                private_rollups.get("expected_product_count")
+                or dev_prod_summary.get("expected_product_count")
+            ),
+            "canonical_repo_count": len(
+                {row["canonical_repo"] for row in products if row["canonical_repo"]}
+            ),
+            "ssh_verified_product_repo_count": _int(
+                private_rollups.get("ssh_verified_product_repo_count")
+                or dev_prod_summary.get("ssh_repo_present_count")
+            ),
+            "main_branch_present_product_repo_count": _int(
+                private_rollups.get("main_branch_present_product_repo_count")
+                or dev_prod_summary.get("main_branch_present_count")
+            ),
+            "dev_branch_present_product_repo_count": _int(
+                private_rollups.get("dev_branch_present_product_repo_count")
+                or dev_prod_summary.get("dev_branch_present_count")
+            ),
+            "dev_prod_environment_split_ready_count": _int(
+                private_rollups.get("dev_prod_environment_split_ready_count")
+                or dev_prod_summary.get("environment_split_ready_count")
+            ),
+            "public_visible_product_repo_count": _int(
+                private_rollups.get("public_visible_product_repo_count")
+                or dev_prod_summary.get("public_visible_count")
+            ),
+            "private_or_auth_only_product_repo_count": _int(
+                private_rollups.get("private_or_auth_only_product_repo_count")
+                or dev_prod_summary.get("private_or_auth_only_count")
+            ),
+            "tokenless_http_404_private_or_auth_product_repo_count": _int(
+                private_rollups.get(
+                    "tokenless_http_404_private_or_auth_product_repo_count"
+                )
+                or dev_prod_summary.get("tokenless_http_404_private_or_auth_count")
+            ),
+            "missing_product_repo_count": _int(
+                private_rollups.get("missing_product_repo_count")
+                or dev_prod_summary.get("missing_repo_count")
+            ),
+            "owner_readiness_row_present_count": sum(
+                1 for row in products if row["owner_readiness_row_present"]
+            ),
+            "all_products_have_source_control_ready": all_rows_ready,
+            "all_active_product_repos_have_owner_readiness_row": (
+                private_rollups.get(
+                    "all_active_product_repos_have_gitea_owner_readiness_row"
+                )
+                is True
+            ),
+            "all_expected_product_repos_have_dev_and_main": (
+                private_rollups.get("all_expected_product_repos_have_dev_and_main")
+                is True
+                or dev_prod_summary.get("all_expected_product_repos_have_dev_and_main")
+                is True
+            ),
+            "backup_bundle_ready": bundle_ready,
+            "backup_bundle_expected_repo_count": _int(
+                bundle_summary.get("expected_repo_count")
+            ),
+            "backup_evidence_linked_product_count": sum(
+                1 for row in products if row["backup_evidence_ref"]
+            ),
+            "backup_bundle_verified_product_count": sum(
+                1
+                for row in products
+                if row["backup_evidence_state"] == "bundle_backup_verified"
+            ),
+            "active_blocker_count": len(blocker_ids),
+        },
+        "products": products,
+        "active_blockers": blocker_ids,
+        "next_action": (
+            "keep_product_governance_matrix_as_source_truth_for_KM_RAG_MCP_LOG_wiring"
+            if status_value == "product_governance_matrix_ready"
+            else "repair_missing_product_source_truth_before_governance_rollout"
+        ),
+        "operation_boundaries": {
+            "read_only_api_allowed": True,
+            "github_api_allowed": False,
+            "github_cli_allowed": False,
+            "gitea_repo_write_allowed": False,
+            "gitea_refs_sync_allowed": False,
+            "gitea_visibility_change_allowed": False,
+            "workflow_dispatch_allowed": False,
+            "secret_value_collection_allowed": False,
+            "raw_session_or_sqlite_read_allowed": False,
+        },
+    }
+
+
+def _product_row(
+    row: dict[str, Any],
+    *,
+    bundle_rows: dict[str, dict[str, Any]],
+    bundle_ready: bool,
+    bundle_ref: str,
+) -> dict[str, Any]:
+    repo = str(row.get("gitea_repo") or "")
+    backup_row = bundle_rows.get(repo)
+    blockers = _strings(row.get("blockers"))
+    source_control_ready = (
+        row.get("ssh_refs_readable") is True
+        and row.get("prod_branch_present") is True
+        and row.get("dev_branch_present") is True
+        and row.get("environment_split_ready") is True
+        and row.get("owner_readiness_row_present") is True
+        and not blockers
+    )
+    visibility = str(row.get("visibility_readback") or "unknown")
+    private_or_auth_only = visibility == "private_or_auth_only"
+    if backup_row:
+        backup_state = (
+            "bundle_backup_verified"
+            if bundle_ready and backup_row.get("ok") is True
+            else "bundle_backup_readback_linked"
+        )
+    else:
+        backup_state = "backup_evidence_ref_only"
+    return {
+        "product_id": str(row.get("product_id") or ""),
+        "status": "ready" if source_control_ready else "action_required",
+        "canonical_repo": repo,
+        "source_control_authority": "gitea",
+        "prod_branch": str(row.get("prod_environment_branch") or "main"),
+        "prod_sha_short": str(row.get("prod_branch_sha") or "")[:8],
+        "prod_branch_present": row.get("prod_branch_present") is True,
+        "dev_branch": str(row.get("dev_environment_branch") or "dev"),
+        "dev_sha_short": str(row.get("dev_branch_sha") or "")[:8],
+        "dev_branch_present": row.get("dev_branch_present") is True,
+        "ssh_refs_readable": row.get("ssh_refs_readable") is True,
+        "environment_split_ready": row.get("environment_split_ready") is True,
+        "owner_readiness_row_present": (
+            row.get("owner_readiness_row_present") is True
+        ),
+        "visibility_readback": visibility,
+        "public_ui_visible": row.get("public_ui_visible") is True,
+        "private_or_auth_only": private_or_auth_only,
+        "tokenless_http_status": _int(row.get("tokenless_http_status")),
+        "tokenless_404_is_visibility_auth": (
+            private_or_auth_only and _int(row.get("tokenless_http_status")) == 404
+        ),
+        "backup_evidence_ref": f"{bundle_ref}#repo={repo}",
+        "backup_evidence_state": backup_state,
+        "source_control_ready": source_control_ready,
+        "blockers": blockers,
+        "next_gate": str(row.get("next_gate") or ""),
+    }
+
+
+def _bundle_repo_index(payload: dict[str, Any] | None) -> dict[str, dict[str, Any]]:
+    rows = _list((payload or {}).get("repos"))
+    indexed: dict[str, dict[str, Any]] = {}
+    for row in rows:
+        repo = str(row.get("repo") or "")
+        if not repo:
+            continue
+        indexed[repo] = row
+    return indexed
+
+
+def _load_json_object(path: Path) -> dict[str, Any]:
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path}: expected JSON object")
+    return payload
+
+
+def _dict(value: Any) -> dict[str, Any]:
+    return value if isinstance(value, dict) else {}
+
+
+def _list(value: Any) -> list[dict[str, Any]]:
+    return [row for row in value if isinstance(row, dict)] if isinstance(value, list) else []
+
+
+def _strings(value: Any) -> list[str]:
+    return [str(item) for item in value if str(item)] if isinstance(value, list) else []
+
+
+def _int(value: Any) -> int:
+    if isinstance(value, bool):
+        return int(value)
+    if isinstance(value, int | float):
+        return int(value)
+    if isinstance(value, str):
+        try:
+            return int(float(value))
+        except ValueError:
+            return 0
+    return 0