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

feat(delivery): expose production deploy readback blocker

Browse Source
This commit is contained in:
Your Name
2026-06-28 19:32:36 +08:00
parent b1da5cc7d3
commit 40dc4315f2
5 changed files with 323 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
apps/api/src/services/awoooi_production_deploy_readback_blocker.py Normal file
View File

@@ -0,0 +1,104 @@
"""AWOOOI production deploy readback blocker.
Loads the committed production deploy/readback blocker snapshot. This service is
read-only: it does not call Gitea, trigger workflows, inspect credentials, touch
K8s, or modify runtime 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__))
_SNAPSHOT_FILE = "awoooi-production-deploy-readback-blocker.snapshot.json"
_SCHEMA_VERSION = "awoooi_production_deploy_readback_blocker_v1"
def load_latest_awoooi_production_deploy_readback_blocker(
operations_dir: Path | None = None,
) -> dict[str, Any]:
"""Load the committed production deploy/readback blocker snapshot."""
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")
_require_schema(payload, str(path))
_require_operation_boundaries(payload, str(path))
_require_rollup_consistency(payload, str(path))
_require_no_internal_network_literals(payload, str(path))
return payload
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")
blocked_flags = {
"deploy_trigger_allowed",
"workflow_modification_allowed",
"gitea_api_write_allowed",
"host_write_allowed",
"k8s_write_allowed",
"secret_read_allowed",
"github_api_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:
readback = _dict(payload.get("readback"))
rollups = _dict(payload.get("rollups"))
blockers = _list(payload.get("blockers"))
next_actions = _list(payload.get("next_actions"))
if rollups.get("hard_blocker_count") != len(blockers):
raise ValueError(f"{label}: hard_blocker_count must match blockers")
if rollups.get("next_action_count") != len(next_actions):
raise ValueError(f"{label}: next_action_count must match next_actions")
if rollups.get("production_image_tag_matches_main") is not (
readback.get("production_image_tag_matches_main") is True
):
raise ValueError(
f"{label}: production_image_tag_matches_main must match readback"
)
if rollups.get("authorized_dispatch_channel_ready") is not (
readback.get("authorized_dispatch_channel_ready") is True
):
raise ValueError(
f"{label}: authorized_dispatch_channel_ready must match readback"
)
def _require_no_internal_network_literals(value: Any, label: str) -> None:
text = json.dumps(value, ensure_ascii=False, sort_keys=True)
forbidden = ["192.168.0.", "api.github.com", "github.com/"]
hits = [item for item in forbidden if item in text]
if hits:
raise ValueError(f"{label}: forbidden literal(s) in public readback: {hits}")
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 []

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

@@ -9,6 +9,9 @@ from __future__ import annotations
from typing import Any
from src.services.awoooi_production_deploy_readback_blocker import (
load_latest_awoooi_production_deploy_readback_blocker,
)
from src.services.awoooi_status_cleanup_dashboard import (
load_latest_awoooi_status_cleanup_dashboard,
)
@@ -28,12 +31,14 @@ _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()
production_deploy = load_latest_awoooi_production_deploy_readback_blocker()
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,
production_deploy=production_deploy,
github=github,
gitea=gitea,
runtime=runtime,
@@ -44,6 +49,7 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
def build_delivery_closure_workbench(
*,
status_cleanup: dict[str, Any],
production_deploy: dict[str, Any],
github: dict[str, Any],
gitea: dict[str, Any],
runtime: dict[str, Any],
@@ -59,6 +65,8 @@ def build_delivery_closure_workbench(
github_preflight.get("internal_governance_writeback")
or github.get("internal_governance_writeback")
)
production_deploy_readback = _dict(production_deploy.get("readback"))
production_deploy_rollups = _dict(production_deploy.get("rollups"))
gitea_status = _dict(gitea.get("program_status"))
gitea_rollups = _dict(gitea.get("rollups"))
runtime_status = _dict(runtime.get("program_status"))
@@ -97,6 +105,46 @@ def build_delivery_closure_workbench(
"href": "/governance?tab=automation-inventory",
"next_action": _first_string(status_cleanup.get("next_actions")),
},
{
"id": "production_deploy",
"source_id": "production_deploy_readback",
"completion_percent": _percent(
100
if production_deploy_rollups.get("production_image_tag_matches_main")
is True
else 40
),
"status": str(production_deploy.get("status") or "unknown"),
"blocker_count": _int(production_deploy_rollups.get("hard_blocker_count")),
"metric": {
"kind": "deploy_readback",
"source_control_main_short_sha": str(
production_deploy_readback.get("source_control_main_short_sha")
or ""
),
"production_image_tag_short_sha": str(
production_deploy_readback.get("production_image_tag_short_sha")
or ""
),
"production_image_tag_matches_main": production_deploy_readback.get(
"production_image_tag_matches_main"
)
is True,
"current_main_cd_run_visible": production_deploy_readback.get(
"current_main_cd_run_visible"
)
is True,
"authorized_dispatch_channel_ready": production_deploy_readback.get(
"authorized_dispatch_channel_ready"
)
is True,
"latest_visible_cd_run_id": str(
production_deploy_readback.get("latest_visible_cd_run_id") or ""
),
},
"href": "/deployments",
"next_action": _first_string(production_deploy.get("next_actions")),
},
{
"id": "github",
"source_id": "github_private_backup",
@@ -208,6 +256,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_ci_cd", gitea),
_source_status("runtime_surface", runtime),
@@ -257,6 +306,29 @@ def build_delivery_closure_workbench(
"workflow_trigger_authorized"
)
is True,
"production_deploy_status": str(production_deploy.get("status") or ""),
"production_deploy_source_control_main_ready": production_deploy_rollups.get(
"source_control_main_ready"
)
is True,
"production_deploy_image_tag_matches_main": production_deploy_rollups.get(
"production_image_tag_matches_main"
)
is True,
"production_deploy_governance_fields_present": production_deploy_rollups.get(
"production_governance_fields_present"
)
is True,
"production_deploy_authorized_dispatch_channel_ready": (
production_deploy_rollups.get("authorized_dispatch_channel_ready")
is True
),
"production_deploy_hard_blocker_count": _int(
production_deploy_rollups.get("hard_blocker_count")
),
"production_deploy_latest_visible_cd_run_id": str(
production_deploy_readback.get("latest_visible_cd_run_id") or ""
),
"github_write_channel_ready": github_preflight.get(
"github_write_channel_ready"
)
@@ -321,6 +393,10 @@ def build_delivery_closure_workbench(
"workflow_trigger_allowed"
)
is True,
"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"
)