wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

feat(governance): expose gitea private inventory p0 readback

Browse Source
This commit is contained in:
Your Name
2026-06-29 15:03:23 +08:00
parent 7ff959b6a8
commit 1dc1550d38
13 changed files with 963 additions and 850 deletions
Download Patch File Download Diff File Expand all files Collapse all files

330
apps/api/src/services/delivery_closure_workbench.py
View File

@@ -18,6 +18,9 @@ from src.services.awoooi_status_cleanup_dashboard import (
from src.services.backup_dr_readiness_matrix import (
load_latest_backup_dr_readiness_matrix,
)
from src.services.gitea_private_inventory_p0_scorecard import (
load_latest_gitea_private_inventory_p0_scorecard,
)
from src.services.gitea_workflow_runner_health import (
load_latest_gitea_workflow_runner_health,
)
@@ -35,7 +38,7 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
"""Load existing delivery snapshots and return a compact workbench model."""
status_cleanup = load_latest_awoooi_status_cleanup_dashboard()
production_deploy = load_latest_awoooi_production_deploy_readback_blocker()
github = _load_github_private_backup_evidence_gate()
private_inventory = load_latest_gitea_private_inventory_p0_scorecard()
cicd_baseline = load_latest_p0_cicd_baseline_source_readiness()
gitea = load_latest_gitea_workflow_runner_health()
runtime = load_latest_runtime_surface_inventory()
@@ -43,7 +46,7 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
return build_delivery_closure_workbench(
status_cleanup=status_cleanup,
production_deploy=production_deploy,
github=github,
private_inventory=private_inventory,
cicd_baseline=cicd_baseline,
gitea=gitea,
runtime=runtime,
@@ -55,7 +58,7 @@ def build_delivery_closure_workbench(
*,
status_cleanup: dict[str, Any],
production_deploy: dict[str, Any],
github: dict[str, Any],
private_inventory: dict[str, Any],
cicd_baseline: dict[str, Any],
gitea: dict[str, Any],
runtime: dict[str, Any],
@@ -63,13 +66,11 @@ def build_delivery_closure_workbench(
) -> dict[str, Any]:
"""Build the delivery workbench response from already validated snapshots."""
status_summary = _dict(status_cleanup.get("summary"))
github_summary = _dict(github.get("summary"))
github_boundaries = _dict(github.get("operation_boundaries"))
github_preflight = _dict(github.get("controlled_execution_preflight"))
github_operator_unblock = _dict(github_preflight.get("operator_unblock"))
github_governance_writeback = _dict(
github_preflight.get("internal_governance_writeback")
or github.get("internal_governance_writeback")
private_inventory_readback = _dict(private_inventory.get("readback"))
private_inventory_rollups = _dict(private_inventory.get("rollups"))
private_inventory_boundaries = _dict(private_inventory.get("operation_boundaries"))
private_inventory_retired_github = _dict(
private_inventory.get("github_retired_context")
)
cicd_baseline_readback = _dict(cicd_baseline.get("readback"))
cicd_baseline_rollups = _dict(cicd_baseline.get("rollups"))
@@ -82,14 +83,8 @@ def build_delivery_closure_workbench(
backup_status = _dict(backup.get("program_status"))
backup_rollups = _dict(backup.get("rollups"))
github_required = _int(github_summary.get("approval_required_target_count"))
github_verified = _int(github_summary.get("private_backup_verified_count"))
github_preflight_blockers = _int(
github_preflight.get("blocked_preflight_target_count")
)
github_lane_blockers = max(
_int(github_summary.get("blocked_target_count")),
github_preflight_blockers,
private_inventory_blockers = _int(
private_inventory_rollups.get("active_blocker_count")
)
runtime_action_required = set(
_strings(runtime_rollups.get("action_required_surface_ids"))
@@ -364,59 +359,60 @@ def build_delivery_closure_workbench(
"next_action": _first_string(production_deploy.get("next_actions")),
},
{
"id": "github",
"source_id": "github_private_backup",
"id": "gitea_private_inventory",
"source_id": "gitea_private_inventory_p0_scorecard",
"completion_percent": _percent(
(github_verified / github_required) * 100 if github_required else 0
private_inventory_rollups.get("review_readiness_percent")
),
"status": str(
github_preflight.get("status") or github.get("status") or "unknown"
),
"blocker_count": github_lane_blockers,
"status": str(private_inventory.get("status") or "unknown"),
"blocker_count": private_inventory_blockers,
"metric": {
"kind": "private_backup_verified",
"verified": github_verified,
"total": github_required,
"controlled_apply_ready": _int(
github_preflight.get("controlled_apply_ready_count")
"kind": "private_inventory",
"workplan_id": str(
private_inventory_readback.get("workplan_id") or "P0-003"
),
"blocked_preflight": github_preflight_blockers,
"write_channel_ready": github_preflight.get(
"github_write_channel_ready"
)
is True,
"github_account_status": str(
github_preflight.get("github_account_status") or "unknown"
"private_inventory_source": str(
private_inventory_readback.get("private_inventory_source")
or "gitea"
),
"github_account_suspended": github_preflight.get(
"github_account_suspended"
)
is True,
"internal_governance_ready": github_governance_writeback.get("ready")
is True,
"km_writeback_ready": _int(
github_governance_writeback.get("km_writeback_ready_count")
"gitea_repo_inventory_status": str(
private_inventory_rollups.get("gitea_repo_inventory_status")
or "unknown"
),
"playbook_writeback_ready": _int(
github_governance_writeback.get(
"playbook_writeback_ready_count"
"gitea_visibility_scope": str(
private_inventory_rollups.get("gitea_visibility_scope")
or "unknown"
),
"gitea_public_repo_count": _int(
private_inventory_rollups.get("gitea_public_repo_count")
),
"expected_product_count": _int(
private_inventory_rollups.get("expected_product_count")
),
"present_product_row_count": _int(
private_inventory_rollups.get("present_product_row_count")
),
"missing_product_row_count": _int(
private_inventory_rollups.get("missing_product_row_count")
),
"accepted_inventory_payload_count": _int(
private_inventory_rollups.get("accepted_inventory_payload_count")
),
"owner_coverage_attestation_received_count": _int(
private_inventory_rollups.get(
"owner_coverage_attestation_received_count"
)
),
"github_lane_excluded_from_p0_blocker_count": (
private_inventory_rollups.get(
"github_lane_excluded_from_p0_blocker_count"
)
is True
),
"active_blockers": _strings(private_inventory.get("active_blockers")),
},
"href": "/governance?tab=automation-inventory",
"operator_unblock": github_operator_unblock,
"internal_governance_writeback": github_governance_writeback,
"next_action": str(
_first_string(github_operator_unblock.get("required_actions"))
or github_operator_unblock.get("safe_handoff")
or github_preflight.get("operator_unblock_status")
or ""
)
or str(
_first_target_action(github_preflight.get("targets"))
or github.get("next_action")
or _first_target_action(github.get("targets"))
),
"href": "/delivery",
"next_action": str(private_inventory_readback.get("safe_next_step") or ""),
},
{
"id": "cicd_baseline",
@@ -573,7 +569,7 @@ def build_delivery_closure_workbench(
source_statuses = [
_source_status("status_cleanup", status_cleanup),
_source_status("production_deploy_readback", production_deploy),
_source_status("github_private_backup", github),
_source_status("gitea_private_inventory_p0_scorecard", private_inventory),
_source_status("p0_cicd_baseline_source_readiness", cicd_baseline),
_source_status("gitea_ci_cd", gitea),
_source_status("runtime_surface", runtime),
@@ -610,19 +606,68 @@ def build_delivery_closure_workbench(
"average_completion_percent": average_completion,
"high_risk_blocker_count": high_risk_blocker_count,
"runtime_execution_authorized": False,
"remote_write_authorized": github_boundaries.get("github_api_write_allowed")
is True,
"repo_creation_authorized": github_summary.get("repo_creation_authorized")
is True,
"visibility_change_authorized": github_summary.get(
"visibility_change_authorized"
)
is True,
"refs_sync_authorized": github_summary.get("refs_sync_authorized") is True,
"workflow_trigger_authorized": github_summary.get(
"workflow_trigger_authorized"
)
is True,
"remote_write_authorized": False,
"repo_creation_authorized": False,
"visibility_change_authorized": False,
"refs_sync_authorized": False,
"workflow_trigger_authorized": False,
"gitea_private_inventory_status": str(private_inventory.get("status") or ""),
"gitea_private_inventory_workplan_id": str(
private_inventory_readback.get("workplan_id") or ""
),
"gitea_private_inventory_source": str(
private_inventory_readback.get("private_inventory_source") or ""
),
"gitea_private_inventory_review_readiness_percent": _int(
private_inventory_rollups.get("review_readiness_percent")
),
"gitea_private_inventory_active_blocker_count": private_inventory_blockers,
"gitea_private_inventory_repo_inventory_status": str(
private_inventory_rollups.get("gitea_repo_inventory_status") or ""
),
"gitea_private_inventory_visibility_scope": str(
private_inventory_rollups.get("gitea_visibility_scope") or ""
),
"gitea_private_inventory_public_repo_count": _int(
private_inventory_rollups.get("gitea_public_repo_count")
),
"gitea_private_inventory_expected_product_count": _int(
private_inventory_rollups.get("expected_product_count")
),
"gitea_private_inventory_present_product_row_count": _int(
private_inventory_rollups.get("present_product_row_count")
),
"gitea_private_inventory_missing_product_row_count": _int(
private_inventory_rollups.get("missing_product_row_count")
),
"gitea_private_inventory_accepted_payload_count": _int(
private_inventory_rollups.get("accepted_inventory_payload_count")
),
"gitea_private_inventory_owner_coverage_attestation_received_count": _int(
private_inventory_rollups.get(
"owner_coverage_attestation_received_count"
)
),
"gitea_private_inventory_all_active_product_repos_have_owner_readiness_row": (
private_inventory_rollups.get(
"all_active_product_repos_have_gitea_owner_readiness_row"
)
is True
),
"gitea_private_inventory_safe_next_step": str(
private_inventory_readback.get("safe_next_step") or ""
),
"github_global_freeze_enabled": True,
"github_lane_status": str(
private_inventory_retired_github.get("status")
or "stopped_retired_do_not_use"
),
"github_lane_excluded_from_p0_blocker_count": (
private_inventory_rollups.get(
"github_lane_excluded_from_p0_blocker_count"
)
is True
),
"p0_cicd_baseline_status": str(cicd_baseline.get("status") or ""),
"p0_cicd_baseline_workplan_id": str(
cicd_baseline_readback.get("workplan_id") or ""
@@ -910,49 +955,21 @@ def build_delivery_closure_workbench(
"backup_credential_escrow_forbidden_true_field_count": _int(
backup_rollups.get("credential_escrow_forbidden_true_field_count")
),
"github_write_channel_ready": github_preflight.get(
"github_write_channel_ready"
)
is True,
"github_account_status": str(
github_preflight.get("github_account_status") or "unknown"
),
"github_account_suspended": github_preflight.get("github_account_suspended")
is True,
"github_api_forbidden_count": _int(
github_preflight.get("github_api_forbidden_count")
),
"github_controlled_apply_ready_count": _int(
github_preflight.get("controlled_apply_ready_count")
),
"github_blocked_preflight_target_count": github_preflight_blockers,
"github_operator_unblock_required": github_operator_unblock.get("required")
is True,
"github_operator_unblock_status": str(
github_operator_unblock.get("status") or ""
),
"github_internal_governance_writeback_ready": github_governance_writeback.get(
"ready"
)
is True,
"github_mcp_evidence_packet_count": _int(
github_governance_writeback.get("mcp_evidence_packet_count")
),
"github_rag_evidence_ref_count": _int(
github_governance_writeback.get("rag_evidence_ref_count")
),
"github_km_writeback_ready_count": _int(
github_governance_writeback.get("km_writeback_ready_count")
),
"github_playbook_writeback_ready_count": _int(
github_governance_writeback.get("playbook_writeback_ready_count")
),
"github_timeline_log_ready_count": _int(
github_governance_writeback.get("timeline_log_ready_count")
),
"github_logbook_entry_count": _int(
github_governance_writeback.get("logbook_entry_count")
),
"github_write_channel_ready": False,
"github_account_status": "stopped_retired_do_not_use",
"github_account_suspended": False,
"github_api_forbidden_count": 0,
"github_controlled_apply_ready_count": 0,
"github_blocked_preflight_target_count": 0,
"github_operator_unblock_required": False,
"github_operator_unblock_status": "github_global_freeze_stopped_retired_do_not_use",
"github_internal_governance_writeback_ready": False,
"github_mcp_evidence_packet_count": 0,
"github_rag_evidence_ref_count": 0,
"github_km_writeback_ready_count": 0,
"github_playbook_writeback_ready_count": 0,
"github_timeline_log_ready_count": 0,
"github_logbook_entry_count": 0,
"secret_values_collected": False,
},
"source_statuses": source_statuses,
@@ -961,29 +978,27 @@ def build_delivery_closure_workbench(
"operation_boundaries": {
"read_only_api_allowed": True,
"runtime_write_allowed": False,
"remote_write_allowed": github_boundaries.get("github_api_write_allowed")
is True,
"repo_creation_allowed": github_boundaries.get("repo_creation_allowed")
is True,
"visibility_change_allowed": github_boundaries.get(
"visibility_change_allowed"
)
is True,
"refs_sync_allowed": github_boundaries.get("refs_sync_allowed") is True,
"workflow_trigger_allowed": github_boundaries.get(
"workflow_trigger_allowed"
)
is True,
"remote_write_allowed": False,
"repo_creation_allowed": False,
"visibility_change_allowed": False,
"refs_sync_allowed": False,
"workflow_trigger_allowed": False,
"production_deploy_trigger_allowed": _dict(
production_deploy.get("operation_boundaries")
).get("deploy_trigger_allowed")
is True,
"github_write_channel_ready": github_preflight.get(
"github_write_channel_ready"
"gitea_api_write_allowed": private_inventory_boundaries.get(
"gitea_api_write_allowed"
)
is True,
"github_controlled_apply_allowed": github_preflight.get("preflight_ready")
is True,
"gitea_authenticated_inventory_import_execution_allowed": (
private_inventory_boundaries.get(
"authenticated_inventory_import_execution_allowed"
)
is True
),
"github_write_channel_ready": False,
"github_controlled_apply_allowed": False,
"secret_value_collection_allowed": False,
"backup_restore_execution_allowed": False,
"active_scan_allowed": False,
@@ -1004,40 +1019,6 @@ def _source_status(source_id: str, payload: dict[str, Any]) -> dict[str, Any]:
}
def _load_github_private_backup_evidence_gate() -> dict[str, Any]:
try:
from src.services.github_target_private_backup_evidence_gate import (
load_latest_github_target_private_backup_evidence_gate,
)
return load_latest_github_target_private_backup_evidence_gate()
except ModuleNotFoundError as exc:
if exc.name != "src.services.github_target_private_backup_evidence_gate":
raise
return _missing_github_private_backup_source(
"service_module_missing_on_release_base"
)
except FileNotFoundError:
return _missing_github_private_backup_source("snapshot_missing_on_release_base")
def _missing_github_private_backup_source(reason: str) -> dict[str, Any]:
return {
"schema_version": "missing_github_target_private_backup_evidence_gate_v1",
"generated_at": "",
"source_missing": True,
"missing_reason": reason,
"status": "blocked_github_private_backup_source_missing",
"next_action": "補上 GitHub 私有備援 evidence gate source 後重新回讀，不可把缺席來源標成已驗證。",
"summary": {
"approval_required_target_count": 0,
"private_backup_verified_count": 0,
"blocked_target_count": 1,
},
"targets": [],
}
def _tone(blocker_count: int, percent: int) -> str:
if blocker_count > 0:
return "danger"
@@ -1074,15 +1055,6 @@ def _first_string(value: Any) -> str:
return ""
def _first_target_action(value: Any) -> str:
if not isinstance(value, list):
return ""
for row in value:
if isinstance(row, dict) and row.get("approval_required") is True:
return str(row.get("next_action") or "")
return _first_row_action(value)
def _first_contract_action(value: Any) -> str:
if not isinstance(value, list):
return ""

282
apps/api/src/services/gitea_private_inventory_p0_scorecard.py Normal file
View File

@@ -0,0 +1,282 @@
"""P0-003 Gitea-only private inventory scorecard readback.
This loader promotes the committed Gitea private inventory scorecard into a
stable API contract. It does not call Gitea or GitHub, does not collect token
values, and does not write repository refs or workflow state.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from src.services.snapshot_paths import default_operations_dir
_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
_SCORECARD_FILE = "awoooi-gitea-private-inventory-p0-scorecard.snapshot.json"
_SOURCE_SCHEMA_VERSION = "awoooi_gitea_private_inventory_p0_scorecard_v1"
_API_SCHEMA_VERSION = "gitea_private_inventory_p0_scorecard_readback_v1"
def load_latest_gitea_private_inventory_p0_scorecard(
operations_dir: Path | None = None,
) -> dict[str, Any]:
"""Load and validate the committed P0-003 Gitea-only scorecard."""
directory = operations_dir or _DEFAULT_OPERATIONS_DIR
path = directory / _SCORECARD_FILE
with path.open(encoding="utf-8") as handle:
scorecard = json.load(handle)
if not isinstance(scorecard, dict):
raise ValueError(f"{path}: expected JSON object")
_require_source_scorecard(scorecard, str(path))
payload = _build_payload(scorecard, path)
_require_operation_boundaries(payload, str(path))
_require_rollup_consistency(payload, str(path))
return payload
def _build_payload(scorecard: dict[str, Any], path: Path) -> dict[str, Any]:
gitea_inventory = _dict(scorecard.get("gitea_inventory"))
import_acceptance = _dict(scorecard.get("authenticated_import_acceptance"))
coverage_attestation = _dict(scorecard.get("coverage_attestation"))
product_coverage = _dict(scorecard.get("product_row_coverage"))
active_blockers = _strings(scorecard.get("active_blockers"))
exit_criteria = _strings(scorecard.get("exit_criteria"))
met_criteria = _met_exit_criteria(scorecard, gitea_inventory, product_coverage)
review_readiness_percent = _percent(
(len(met_criteria) / max(len(exit_criteria), 1)) * 100
)
return {
"schema_version": _API_SCHEMA_VERSION,
"generated_at": str(scorecard.get("generated_at") or ""),
"priority": "P0-003",
"scope": "gitea_private_inventory_p0_scorecard",
"status": str(scorecard.get("status") or "unknown"),
"readback": {
"workplan_id": str(scorecard.get("workplan_id") or "P0-003"),
"workplan_title": "取得 Gitea private inventory 權限",
"source_scorecard_ref": f"docs/operations/{path.name}",
"source_control_authority": str(
scorecard.get("source_control_authority") or "gitea"
),
"private_inventory_source": str(
scorecard.get("private_inventory_source") or "gitea"
),
"safe_next_step": str(scorecard.get("safe_next_step") or ""),
},
"gitea_inventory": gitea_inventory,
"authenticated_import_acceptance": import_acceptance,
"coverage_attestation": coverage_attestation,
"product_row_coverage": product_coverage,
"active_blockers": active_blockers,
"exit_criteria": exit_criteria,
"met_exit_criteria": met_criteria,
"rollups": {
"active_blocker_count": len(active_blockers),
"review_readiness_percent": review_readiness_percent,
"gitea_repo_inventory_status": str(
gitea_inventory.get("status") or "unknown"
),
"gitea_visibility_scope": str(
gitea_inventory.get("visibility_scope") or "unknown"
),
"gitea_public_repo_count": _int(gitea_inventory.get("repo_count")),
"accepted_inventory_payload_count": _int(
import_acceptance.get("accepted_payload_count")
),
"owner_coverage_attestation_received_count": _int(
coverage_attestation.get("received_attestation_count")
),
"owner_coverage_attestation_accepted_count": _int(
coverage_attestation.get("accepted_attestation_count")
),
"expected_product_count": _int(
product_coverage.get("expected_product_count")
),
"present_product_row_count": _int(
product_coverage.get("present_product_row_count")
),
"missing_product_row_count": _int(
product_coverage.get("missing_product_row_count")
),
"ready_product_count": _int(product_coverage.get("ready_product_count")),
"blocked_product_count": _int(
product_coverage.get("blocked_product_count")
),
"internal_or_authenticated_inventory_required_count": _int(
product_coverage.get(
"internal_or_authenticated_inventory_required_count"
)
),
"all_active_product_repos_have_gitea_owner_readiness_row": (
product_coverage.get(
"all_active_product_repos_have_gitea_owner_readiness_row"
)
is True
),
"github_lane_excluded_from_p0_blocker_count": (
scorecard.get("github_lane_excluded_from_p0_blocker_count")
is True
),
"github_api_used": scorecard.get("github_api_used") is True,
"github_cli_used": scorecard.get("github_cli_used") is True,
"github_primary_switch_authorized": (
scorecard.get("github_primary_switch_authorized") is True
),
},
"operation_boundaries": {
"read_only_api_allowed": True,
"gitea_api_write_allowed": False,
"gitea_repo_creation_allowed": False,
"gitea_refs_sync_allowed": False,
"gitea_visibility_change_allowed": False,
"gitea_workflow_trigger_allowed": False,
"authenticated_inventory_import_execution_allowed": False,
"owner_attestation_write_allowed": False,
"token_value_collection_allowed": False,
"secret_value_collection_allowed": False,
"github_api_allowed": False,
"github_cli_allowed": False,
"github_repo_creation_allowed": False,
"github_refs_sync_allowed": False,
"github_primary_switch_allowed": False,
"raw_session_or_sqlite_read_allowed": False,
},
"github_retired_context": {
"status": str(scorecard.get("github_status") or "stopped_retired_do_not_use"),
"excluded_from_active_p0_blocker_math": (
scorecard.get("github_lane_excluded_from_p0_blocker_count")
is True
),
"api_used": scorecard.get("github_api_used") is True,
"cli_used": scorecard.get("github_cli_used") is True,
"primary_switch_authorized": (
scorecard.get("github_primary_switch_authorized") is True
),
},
}
def _require_source_scorecard(payload: dict[str, Any], label: str) -> None:
if payload.get("schema_version") != _SOURCE_SCHEMA_VERSION:
raise ValueError(f"{label}: schema_version must be {_SOURCE_SCHEMA_VERSION}")
if payload.get("source_control_authority") != "gitea":
raise ValueError(f"{label}: source_control_authority must be gitea")
if payload.get("private_inventory_source") != "gitea":
raise ValueError(f"{label}: private_inventory_source must be gitea")
if payload.get("github_status") != "stopped_retired_do_not_use":
raise ValueError(f"{label}: github_status must be stopped_retired_do_not_use")
if payload.get("github_lane_excluded_from_p0_blocker_count") is not True:
raise ValueError(f"{label}: github lane must be excluded from P0 blocker math")
if payload.get("github_api_used") is not False:
raise ValueError(f"{label}: github_api_used must remain false")
if payload.get("github_cli_used") is not False:
raise ValueError(f"{label}: github_cli_used must remain false")
if payload.get("github_primary_switch_authorized") is not False:
raise ValueError(f"{label}: github_primary_switch_authorized must remain false")
import_acceptance = _dict(payload.get("authenticated_import_acceptance"))
if import_acceptance.get("token_value_collection_allowed") is not False:
raise ValueError(f"{label}: token value collection must remain false")
if import_acceptance.get("execution_authorized") is not False:
raise ValueError(f"{label}: import execution must remain false")
coverage_attestation = _dict(payload.get("coverage_attestation"))
if coverage_attestation.get("execution_authorized") is not False:
raise ValueError(f"{label}: attestation write must remain false")
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")
blocked_flags = {
"gitea_api_write_allowed",
"gitea_repo_creation_allowed",
"gitea_refs_sync_allowed",
"gitea_visibility_change_allowed",
"gitea_workflow_trigger_allowed",
"authenticated_inventory_import_execution_allowed",
"owner_attestation_write_allowed",
"token_value_collection_allowed",
"secret_value_collection_allowed",
"github_api_allowed",
"github_cli_allowed",
"github_repo_creation_allowed",
"github_refs_sync_allowed",
"github_primary_switch_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"))
blockers = _strings(payload.get("active_blockers"))
if rollups.get("active_blocker_count") != len(blockers):
raise ValueError(f"{label}: active_blocker_count mismatch")
if rollups.get("github_lane_excluded_from_p0_blocker_count") is not True:
raise ValueError(f"{label}: github lane exclusion must remain true")
forbidden_true = [
"github_api_used",
"github_cli_used",
"github_primary_switch_authorized",
]
unexpected_true = [field for field in forbidden_true if rollups.get(field) is True]
if unexpected_true:
raise ValueError(f"{label}: GitHub forbidden flags true: {unexpected_true}")
if rollups.get("missing_product_row_count") != 0:
raise ValueError(f"{label}: active product readiness rows must remain complete")
if (
rollups.get("all_active_product_repos_have_gitea_owner_readiness_row")
is not True
):
raise ValueError(f"{label}: active product owner readiness rows incomplete")
def _met_exit_criteria(
scorecard: dict[str, Any],
gitea_inventory: dict[str, Any],
product_coverage: dict[str, Any],
) -> list[str]:
met: list[str] = []
if scorecard.get("private_inventory_source") == "gitea":
met.append("private_inventory_source=gitea")
if scorecard.get("github_lane_excluded_from_p0_blocker_count") is True:
met.append("github_lane_excluded_from_p0_blocker_count=true")
if gitea_inventory.get("status") == "ok":
met.append("gitea_repo_inventory.status=ok")
if gitea_inventory.get("visibility_scope") in {"authenticated", "admin_export"}:
met.append("gitea_repo_inventory.visibility_scope in authenticated/admin_export")
if (
product_coverage.get("all_active_product_repos_have_gitea_owner_readiness_row")
is True
):
met.append("all_active_product_repos_have_gitea_owner_readiness_row=true")
return met
def _dict(value: Any) -> dict[str, Any]:
return value if isinstance(value, dict) else {}
def _strings(value: Any) -> list[str]:
if not isinstance(value, list):
return []
return [str(item) for item in value if item is not None]
def _int(value: Any) -> int:
if isinstance(value, bool):
return int(value)
if isinstance(value, int | float):
return int(value)
return 0
def _percent(value: Any) -> int:
return max(0, min(100, round(float(value or 0))))