429 lines
16 KiB
Python
429 lines
16 KiB
Python
"""Delivery closure workbench summary.
|
|
|
|
Builds the product-facing delivery closure view from existing committed,
|
|
read-only snapshots. The summary is intentionally compact so the UI does not
|
|
need to fan out across five separate endpoints or duplicate blocker math.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from src.services.awoooi_status_cleanup_dashboard import (
|
|
load_latest_awoooi_status_cleanup_dashboard,
|
|
)
|
|
from src.services.backup_dr_readiness_matrix import (
|
|
load_latest_backup_dr_readiness_matrix,
|
|
)
|
|
from src.services.gitea_workflow_runner_health import (
|
|
load_latest_gitea_workflow_runner_health,
|
|
)
|
|
from src.services.runtime_surface_inventory import (
|
|
load_latest_runtime_surface_inventory,
|
|
)
|
|
|
|
_SCHEMA_VERSION = "delivery_closure_workbench_v1"
|
|
|
|
|
|
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()
|
|
github = _load_github_private_backup_evidence_gate()
|
|
gitea = load_latest_gitea_workflow_runner_health()
|
|
runtime = load_latest_runtime_surface_inventory()
|
|
backup = load_latest_backup_dr_readiness_matrix()
|
|
return build_delivery_closure_workbench(
|
|
status_cleanup=status_cleanup,
|
|
github=github,
|
|
gitea=gitea,
|
|
runtime=runtime,
|
|
backup=backup,
|
|
)
|
|
|
|
|
|
def build_delivery_closure_workbench(
|
|
*,
|
|
status_cleanup: dict[str, Any],
|
|
github: dict[str, Any],
|
|
gitea: dict[str, Any],
|
|
runtime: dict[str, Any],
|
|
backup: dict[str, Any],
|
|
) -> 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"))
|
|
gitea_status = _dict(gitea.get("program_status"))
|
|
gitea_rollups = _dict(gitea.get("rollups"))
|
|
runtime_status = _dict(runtime.get("program_status"))
|
|
runtime_rollups = _dict(runtime.get("rollups"))
|
|
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,
|
|
)
|
|
runtime_action_required = set(
|
|
_strings(runtime_rollups.get("action_required_surface_ids"))
|
|
)
|
|
runtime_secret_surfaces = set(_strings(runtime_rollups.get("secret_surface_ids")))
|
|
|
|
lanes = [
|
|
{
|
|
"id": "release",
|
|
"source_id": "status_cleanup",
|
|
"completion_percent": _percent(
|
|
status_summary.get("overall_completion_percent")
|
|
),
|
|
"status": str(status_summary.get("dashboard_status") or "unknown"),
|
|
"blocker_count": _int(status_summary.get("blocked_gate_count")),
|
|
"metric": {
|
|
"kind": "blocked_gate",
|
|
"blocked": _int(status_summary.get("blocked_gate_count")),
|
|
"total": _int(status_summary.get("gate_count")),
|
|
},
|
|
"href": "/governance?tab=automation-inventory",
|
|
"next_action": _first_string(status_cleanup.get("next_actions")),
|
|
},
|
|
{
|
|
"id": "github",
|
|
"source_id": "github_private_backup",
|
|
"completion_percent": _percent(
|
|
(github_verified / github_required) * 100 if github_required else 0
|
|
),
|
|
"status": str(
|
|
github_preflight.get("status") or github.get("status") or "unknown"
|
|
),
|
|
"blocker_count": github_lane_blockers,
|
|
"metric": {
|
|
"kind": "private_backup_verified",
|
|
"verified": github_verified,
|
|
"total": github_required,
|
|
"controlled_apply_ready": _int(
|
|
github_preflight.get("controlled_apply_ready_count")
|
|
),
|
|
"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"
|
|
),
|
|
"github_account_suspended": github_preflight.get(
|
|
"github_account_suspended"
|
|
)
|
|
is True,
|
|
},
|
|
"href": "/governance?tab=automation-inventory",
|
|
"operator_unblock": github_operator_unblock,
|
|
"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"))
|
|
),
|
|
},
|
|
{
|
|
"id": "gitea",
|
|
"source_id": "gitea_ci_cd",
|
|
"completion_percent": _percent(
|
|
gitea_status.get("overall_completion_percent")
|
|
),
|
|
"status": str(gitea_status.get("current_task_id") or "unknown"),
|
|
"blocker_count": len(
|
|
_strings(gitea_rollups.get("runner_contracts_requiring_action"))
|
|
),
|
|
"metric": {
|
|
"kind": "workflow_count",
|
|
"count": _int(gitea_rollups.get("total_workflows")),
|
|
},
|
|
"href": "/deployments",
|
|
"next_action": _first_contract_action(gitea.get("runner_contracts")),
|
|
},
|
|
{
|
|
"id": "runtime",
|
|
"source_id": "runtime_surface",
|
|
"completion_percent": _percent(
|
|
runtime_status.get("overall_completion_percent")
|
|
),
|
|
"status": str(runtime_status.get("current_task_id") or "unknown"),
|
|
"blocker_count": len(runtime_action_required | runtime_secret_surfaces),
|
|
"metric": {
|
|
"kind": "surface_count",
|
|
"total": _int(runtime_rollups.get("total_surfaces")),
|
|
},
|
|
"href": "/governance?tab=automation-inventory",
|
|
"next_action": _first_surface_action(runtime.get("runtime_surfaces")),
|
|
},
|
|
{
|
|
"id": "backup",
|
|
"source_id": "backup_dr",
|
|
"completion_percent": _percent(
|
|
backup_status.get("overall_completion_percent")
|
|
),
|
|
"status": str(backup_status.get("current_task_id") or "unknown"),
|
|
"blocker_count": len(_strings(backup_rollups.get("blocked_row_ids"))),
|
|
"metric": {
|
|
"kind": "readiness_row_count",
|
|
"rows": _int(backup_rollups.get("total_rows")),
|
|
},
|
|
"href": "/operations",
|
|
"next_action": _first_backup_action(backup.get("readiness_rows")),
|
|
},
|
|
]
|
|
|
|
for lane in lanes:
|
|
lane["tone"] = _tone(
|
|
_int(lane["blocker_count"]), _int(lane["completion_percent"])
|
|
)
|
|
|
|
source_statuses = [
|
|
_source_status("status_cleanup", status_cleanup),
|
|
_source_status("github_private_backup", github),
|
|
_source_status("gitea_ci_cd", gitea),
|
|
_source_status("runtime_surface", runtime),
|
|
_source_status("backup_dr", backup),
|
|
]
|
|
generated_candidates = [
|
|
source["generated_at"] for source in source_statuses if source["generated_at"]
|
|
]
|
|
loaded_source_count = sum(1 for source in source_statuses if source["loaded"])
|
|
high_risk_blocker_count = sum(_int(lane["blocker_count"]) for lane in lanes)
|
|
average_completion = _percent(
|
|
sum(_int(lane["completion_percent"]) for lane in lanes) / max(len(lanes), 1)
|
|
)
|
|
next_focus = [
|
|
{
|
|
"lane_id": lane["id"],
|
|
"blocker_count": lane["blocker_count"],
|
|
"completion_percent": lane["completion_percent"],
|
|
"next_action": lane["next_action"],
|
|
}
|
|
for lane in lanes
|
|
if _int(lane["blocker_count"]) > 0 or _int(lane["completion_percent"]) < 80
|
|
][:5]
|
|
|
|
return {
|
|
"schema_version": _SCHEMA_VERSION,
|
|
"generated_at": max(generated_candidates) if generated_candidates else "",
|
|
"status": "blocked_delivery_actions_required"
|
|
if high_risk_blocker_count
|
|
else "ready",
|
|
"summary": {
|
|
"source_count": len(source_statuses),
|
|
"loaded_source_count": loaded_source_count,
|
|
"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,
|
|
"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 ""
|
|
),
|
|
"secret_values_collected": False,
|
|
},
|
|
"source_statuses": source_statuses,
|
|
"lanes": lanes,
|
|
"next_focus": next_focus,
|
|
"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,
|
|
"github_write_channel_ready": github_preflight.get(
|
|
"github_write_channel_ready"
|
|
)
|
|
is True,
|
|
"github_controlled_apply_allowed": github_preflight.get("preflight_ready")
|
|
is True,
|
|
"secret_value_collection_allowed": False,
|
|
"backup_restore_execution_allowed": False,
|
|
"active_scan_allowed": False,
|
|
},
|
|
}
|
|
|
|
|
|
def _source_status(source_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
source_missing = payload.get("source_missing") is True
|
|
return {
|
|
"id": source_id,
|
|
"loaded": not source_missing,
|
|
"schema_version": str(payload.get("schema_version") or ""),
|
|
"generated_at": str(payload.get("generated_at") or ""),
|
|
"missing_reason": str(payload.get("missing_reason") or "")
|
|
if source_missing
|
|
else "",
|
|
}
|
|
|
|
|
|
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"
|
|
if percent < 80:
|
|
return "warn"
|
|
return "ok"
|
|
|
|
|
|
def _dict(value: Any) -> dict[str, Any]:
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
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))))
|
|
|
|
|
|
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 _first_string(value: Any) -> str:
|
|
if isinstance(value, list) and value:
|
|
return str(value[0])
|
|
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 ""
|
|
for row in value:
|
|
if isinstance(row, dict) and row.get("status") == "action_required":
|
|
return str(row.get("next_action") or "")
|
|
return _first_row_action(value)
|
|
|
|
|
|
def _first_surface_action(value: Any) -> str:
|
|
if not isinstance(value, list):
|
|
return ""
|
|
for row in value:
|
|
if isinstance(row, dict) and row.get("status") != "manifest_mapped":
|
|
return str(row.get("next_action") or "")
|
|
return _first_row_action(value)
|
|
|
|
|
|
def _first_backup_action(value: Any) -> str:
|
|
if not isinstance(value, list):
|
|
return ""
|
|
for row in value:
|
|
if isinstance(row, dict) and row.get("overall_readiness") in {
|
|
"blocked",
|
|
"action_required",
|
|
}:
|
|
return str(row.get("next_action") or "")
|
|
return _first_row_action(value)
|
|
|
|
|
|
def _first_row_action(value: Any) -> str:
|
|
if not isinstance(value, list):
|
|
return ""
|
|
for row in value:
|
|
if isinstance(row, dict) and row.get("next_action"):
|
|
return str(row["next_action"])
|
|
return ""
|