feat(api): expose warning step runtime enablement gate
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 22s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled

This commit is contained in:
Your Name
2026-06-30 00:59:55 +08:00
parent 5827ef875f
commit 060a82282c
7 changed files with 461 additions and 0 deletions

View File

@@ -317,6 +317,8 @@ jobs:
;;
apps/api/src/services/awoooi_gitea_onboarding_warning_step_template_copy_receipt.py)
;;
apps/api/src/services/awoooi_gitea_onboarding_warning_step_runtime_enablement_gate.py)
;;
apps/api/src/services/awoooi_new_product_onboarding_page_model.py)
;;
apps/api/src/services/awoooi_onboarding_reminder_contract.py)
@@ -561,6 +563,7 @@ jobs:
src/services/awoooi_gitea_onboarding_warning_step_template_copy_apply_gate.py \
src/services/awoooi_gitea_onboarding_warning_step_template_copy_execution_plan.py \
src/services/awoooi_gitea_onboarding_warning_step_template_copy_receipt.py \
src/services/awoooi_gitea_onboarding_warning_step_runtime_enablement_gate.py \
src/services/awoooi_new_product_onboarding_page_model.py \
src/services/awoooi_onboarding_reminder_contract.py \
src/services/awoooi_onboarding_source_contracts.py \

View File

@@ -316,6 +316,9 @@ from src.services.ai_technology_radar_readback import (
from src.services.ai_technology_report_cadence_readback import (
load_latest_ai_technology_report_cadence_readback,
)
from src.services.awoooi_gitea_onboarding_warning_step_runtime_enablement_gate import (
load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate,
)
from src.services.awoooi_gitea_onboarding_warning_step_template_copy_apply_gate import (
load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate,
)
@@ -1195,6 +1198,37 @@ async def get_awoooi_gitea_onboarding_warning_step_template_copy_receipt() -> di
) from exc
@router.get(
"/awoooi-gitea-onboarding-warning-step-runtime-enablement-gate",
response_model=dict[str, Any],
summary="取得 AWOOOI Gitea onboarding warning-step runtime enablement gate",
description=(
"讀取 P0-004 warning-step runtime enablement 的受控 gate"
"此端點只回傳 enablement target selector、source diff、check-mode、"
"rollback 與 post-apply verifier。它不修改 workflow、不觸發 workflow、"
"不建立 repo、不同步 refs、不呼叫 GitHub、不讀 secret、不操作 host / K8s。"
),
)
async def get_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate() -> dict[
str, Any
]:
"""回傳 P0-004 warning-step runtime enablement gate 只讀讀回。"""
try:
payload = await asyncio.to_thread(
load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate
)
return redact_public_lan_topology(payload)
except (json.JSONDecodeError, ValueError) as exc:
logger.error(
"awoooi_gitea_onboarding_warning_step_runtime_enablement_gate_invalid",
error=str(exc),
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="P0-004 warning-step runtime enablement gate 無效",
) from exc
@router.get(
"/reboot-auto-recovery-slo-scorecard",
response_model=dict[str, Any],

View File

@@ -0,0 +1,213 @@
"""AWOOOI warning-step runtime enablement gate readback."""
from __future__ import annotations
from typing import Any
from src.services.awoooi_gitea_onboarding_warning_step_template_copy_receipt import (
load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt,
)
_SCHEMA_VERSION = "awoooi_gitea_onboarding_warning_step_runtime_enablement_gate_v1"
_RECEIPT_READY_STATUS = "controlled_template_copy_receipt_ready"
_WORKFLOW_PATH = ".gitea/workflows/awoooi-onboarding-warning-step.yaml"
def load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate() -> dict[
str, Any
]:
"""Return the controlled gate for enabling the copied warning-step runtime."""
receipt = load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt()
receipt_rollups = _dict(receipt.get("rollups"))
receipt_boundaries = _dict(receipt.get("operation_boundaries"))
receipt_readback = _dict(receipt.get("readback"))
receipt_ready = (
receipt.get("status") == _RECEIPT_READY_STATUS
and receipt.get("active_blockers") == []
and receipt_rollups.get("workflow_file_present") is True
and receipt_rollups.get("workflow_matches_template") is True
and receipt_rollups.get("workflow_dispatch_declared") is True
and receipt_rollups.get("auto_branch_event_count") == 0
and receipt_rollups.get("generic_runner_label_count") == 0
and receipt_rollups.get("fail_closed_execution_switch_present") is True
and receipt_boundaries.get("workflow_trigger_performed") is False
)
active_blockers = _active_blockers(
receipt_ready=receipt_ready,
receipt_rollups=receipt_rollups,
receipt_boundaries=receipt_boundaries,
)
return {
"schema_version": _SCHEMA_VERSION,
"priority": "P0-004",
"scope": "gitea_onboarding_warning_step_runtime_enablement",
"status": (
"controlled_warning_step_runtime_enablement_gate_ready"
if not active_blockers
else "blocked_warning_step_runtime_enablement_gate"
),
"readback": {
"workplan_id": "P0-004-WARNING-STEP-RUNTIME-ENABLEMENT",
"source_receipt_status": receipt.get("status"),
"source_receipt_endpoint": (
"/api/v1/agents/"
"awoooi-gitea-onboarding-warning-step-template-copy-receipt"
),
"workflow_path": receipt_readback.get(
"destination_workflow_path",
_WORKFLOW_PATH,
),
"workflow_content_source": receipt_readback.get("workflow_content_source"),
"workflow_content_sha256_12": receipt_readback.get(
"workflow_content_sha256_12"
),
"runtime_enablement_switch": "AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED",
"current_switch_default": "0",
"desired_switch_value_after_controlled_apply": "1",
"safe_next_step": (
"prepare_controlled_switch_flip_with_pressure_guard_then_verify_no_auto_branch_trigger"
),
},
"controlled_apply": {
"mode": "controlled_apply_prepared",
"controlled_apply_allowed": not active_blockers,
"workflow_runtime_enablement_authorized": not active_blockers,
"workflow_dispatch_authorized": False,
"auto_push_or_pull_request_trigger_authorized": False,
"generic_runner_label_authorized": False,
"runner_pressure_guard_required": True,
"target_selector_required": True,
"source_of_truth_diff_required": True,
"check_mode_required": True,
"rollback_required": True,
"post_apply_verifier_required": True,
},
"target_selector": {
"selector_type": "workflow_runtime_enablement_switch",
"workflow_path": _WORKFLOW_PATH,
"env_key": "AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED",
"expected_current_value": "0",
"desired_value_after_controlled_apply": "1",
"allowed_runner_label": "awoooi-non110-host",
},
"source_of_truth_diff": {
"current_state": "warning_step_runtime_job_fail_closed_disabled",
"desired_state": "warning_step_runtime_job_enabled_for_manual_dispatch_only",
"delta_kind": "env_switch_value_change",
"auto_branch_event_delta_allowed": False,
"generic_runner_label_delta_allowed": False,
"workflow_dispatch_delta_allowed": False,
"raw_secret_payload_included": False,
},
"check_mode": {
"enabled": True,
"required": True,
"checks": [
"verify_template_copy_receipt_ready",
"run_runner_pressure_guard",
"verify_no_push_pull_request_triggers",
"verify_no_generic_runner_labels",
"verify_switch_currently_fail_closed",
"verify_git_diff_contains_only_switch_change",
],
},
"rollback": {
"required": True,
"strategy": "restore_warning_step_execution_switch_to_zero",
"paths": [_WORKFLOW_PATH],
},
"post_apply_verifier": {
"required": True,
"verifier_refs": [
"python3 ops/runner/guard-gitea-runner-pressure.py --root .",
"git diff --check",
(
"pytest apps/api/tests/"
"test_p0_cicd_baseline_source_readiness_api.py"
),
],
"production_readback": (
"/api/v1/agents/"
"awoooi-gitea-onboarding-warning-step-runtime-enablement-gate"
),
},
"rollups": {
"template_copy_receipt_ready": receipt_ready,
"workflow_file_present": receipt_rollups.get("workflow_file_present") is True,
"workflow_matches_template": (
receipt_rollups.get("workflow_matches_template") is True
),
"workflow_dispatch_declared": (
receipt_rollups.get("workflow_dispatch_declared") is True
),
"auto_branch_event_count": _int(receipt_rollups.get("auto_branch_event_count")),
"generic_runner_label_count": _int(
receipt_rollups.get("generic_runner_label_count")
),
"fail_closed_execution_switch_present": (
receipt_rollups.get("fail_closed_execution_switch_present") is True
),
"runner_pressure_guard_required": True,
"controlled_runtime_enablement_ready": not active_blockers,
"active_blocker_count": len(active_blockers),
"workflow_trigger_performed": False,
"runtime_switch_changed": False,
},
"active_blockers": active_blockers,
"operation_boundaries": {
"readback_only": True,
"controlled_runtime_enablement_gate_open": not active_blockers,
"workflow_file_modified_by_this_endpoint": False,
"workflow_trigger_performed": False,
"auto_push_or_pull_request_trigger_allowed": False,
"generic_runner_label_allowed": False,
"github_api_used": False,
"secret_value_collection_allowed": False,
"host_or_k8s_write_performed": False,
"raw_session_or_sqlite_read_allowed": False,
},
}
def _active_blockers(
*,
receipt_ready: bool,
receipt_rollups: dict[str, Any],
receipt_boundaries: dict[str, Any],
) -> list[str]:
blockers: list[str] = []
if not receipt_ready:
blockers.append("template_copy_receipt_not_ready")
if receipt_rollups.get("auto_branch_event_count") != 0:
blockers.append("auto_branch_event_present_in_workflow")
if receipt_rollups.get("generic_runner_label_count") != 0:
blockers.append("generic_runner_label_present_in_workflow")
if receipt_rollups.get("fail_closed_execution_switch_present") is not True:
blockers.append("fail_closed_execution_switch_missing")
if receipt_boundaries.get("workflow_trigger_performed") is not False:
blockers.append("workflow_trigger_already_performed")
return _unique(blockers)
def _dict(value: Any) -> dict[str, Any]:
return value if isinstance(value, dict) else {}
def _int(value: Any) -> int:
try:
return int(value or 0)
except (TypeError, ValueError):
return 0
def _unique(values: list[str]) -> list[str]:
seen = set()
result = []
for value in values:
if value in seen:
continue
seen.add(value)
result.append(value)
return result

View File

@@ -9,6 +9,9 @@ from __future__ import annotations
from typing import Any
from src.services.awoooi_gitea_onboarding_warning_step_runtime_enablement_gate import (
load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate,
)
from src.services.awoooi_gitea_onboarding_warning_step_template_copy_apply_gate import (
load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate,
)
@@ -58,6 +61,9 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
cicd_template_copy_receipt = (
load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt()
)
cicd_runtime_enablement_gate = (
load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate()
)
gitea = load_latest_gitea_workflow_runner_health()
runtime = load_latest_runtime_surface_inventory()
backup = load_latest_backup_dr_readiness_matrix()
@@ -70,6 +76,7 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
cicd_baseline=cicd_baseline,
cicd_template_copy_apply_gate=cicd_template_copy_apply_gate,
cicd_template_copy_receipt=cicd_template_copy_receipt,
cicd_runtime_enablement_gate=cicd_runtime_enablement_gate,
gitea=gitea,
runtime=runtime,
backup=backup,
@@ -86,6 +93,7 @@ def build_delivery_closure_workbench(
cicd_baseline: dict[str, Any],
cicd_template_copy_apply_gate: dict[str, Any],
cicd_template_copy_receipt: dict[str, Any],
cicd_runtime_enablement_gate: dict[str, Any],
gitea: dict[str, Any],
runtime: dict[str, Any],
backup: dict[str, Any],
@@ -121,6 +129,18 @@ def build_delivery_closure_workbench(
cicd_template_copy_receipt_boundaries = _dict(
cicd_template_copy_receipt.get("operation_boundaries")
)
cicd_runtime_enablement_readback = _dict(
cicd_runtime_enablement_gate.get("readback")
)
cicd_runtime_enablement_rollups = _dict(
cicd_runtime_enablement_gate.get("rollups")
)
cicd_runtime_enablement_controlled_apply = _dict(
cicd_runtime_enablement_gate.get("controlled_apply")
)
cicd_runtime_enablement_boundaries = _dict(
cicd_runtime_enablement_gate.get("operation_boundaries")
)
production_deploy_readback = _dict(production_deploy.get("readback"))
production_deploy_rollups = _dict(production_deploy.get("rollups"))
gitea_status = _dict(gitea.get("program_status"))
@@ -873,6 +893,48 @@ def build_delivery_closure_workbench(
)
is True
),
"runtime_enablement_gate_status": str(
cicd_runtime_enablement_gate.get("status") or ""
),
"runtime_enablement_gate_ready": (
cicd_runtime_enablement_gate.get("status")
== "controlled_warning_step_runtime_enablement_gate_ready"
),
"runtime_enablement_gate_active_blocker_count": _int(
cicd_runtime_enablement_rollups.get("active_blocker_count")
),
"runtime_enablement_controlled_apply_allowed": (
cicd_runtime_enablement_controlled_apply.get(
"controlled_apply_allowed"
)
is True
),
"runtime_enablement_workflow_dispatch_authorized": (
cicd_runtime_enablement_controlled_apply.get(
"workflow_dispatch_authorized"
)
is True
),
"runtime_enablement_switch": str(
cicd_runtime_enablement_readback.get("runtime_enablement_switch")
or ""
),
"runtime_enablement_current_switch_default": str(
cicd_runtime_enablement_readback.get("current_switch_default")
or ""
),
"runtime_enablement_desired_switch_value": str(
cicd_runtime_enablement_readback.get(
"desired_switch_value_after_controlled_apply"
)
or ""
),
"runtime_enablement_trigger_performed": (
cicd_runtime_enablement_boundaries.get(
"workflow_trigger_performed"
)
is True
),
"required_source_count": _int(
cicd_baseline_rollups.get("required_source_count")
),
@@ -1313,6 +1375,44 @@ def build_delivery_closure_workbench(
cicd_template_copy_receipt_boundaries.get("workflow_trigger_performed")
is True
),
"p0_cicd_warning_step_runtime_enablement_gate_status": str(
cicd_runtime_enablement_gate.get("status") or ""
),
"p0_cicd_warning_step_runtime_enablement_gate_ready": (
cicd_runtime_enablement_gate.get("status")
== "controlled_warning_step_runtime_enablement_gate_ready"
),
"p0_cicd_warning_step_runtime_enablement_active_blocker_count": _int(
cicd_runtime_enablement_rollups.get("active_blocker_count")
),
"p0_cicd_warning_step_runtime_enablement_controlled_apply_allowed": (
cicd_runtime_enablement_controlled_apply.get(
"controlled_apply_allowed"
)
is True
),
"p0_cicd_warning_step_runtime_enablement_workflow_dispatch_authorized": (
cicd_runtime_enablement_controlled_apply.get(
"workflow_dispatch_authorized"
)
is True
),
"p0_cicd_warning_step_runtime_enablement_switch": str(
cicd_runtime_enablement_readback.get("runtime_enablement_switch") or ""
),
"p0_cicd_warning_step_runtime_enablement_current_switch_default": str(
cicd_runtime_enablement_readback.get("current_switch_default") or ""
),
"p0_cicd_warning_step_runtime_enablement_desired_switch_value": str(
cicd_runtime_enablement_readback.get(
"desired_switch_value_after_controlled_apply"
)
or ""
),
"p0_cicd_warning_step_runtime_enablement_trigger_performed": (
cicd_runtime_enablement_boundaries.get("workflow_trigger_performed")
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"

View File

@@ -371,6 +371,49 @@ def _assert_delivery_workbench_shape(data: dict):
assert data["summary"]["p0_cicd_template_copy_generic_runner_label_count"] == 0
assert data["summary"]["p0_cicd_template_copy_active_workflow_file_created"] is True
assert data["summary"]["p0_cicd_template_copy_workflow_trigger_performed"] is False
assert data["summary"]["p0_cicd_warning_step_runtime_enablement_gate_status"] == (
"controlled_warning_step_runtime_enablement_gate_ready"
)
assert data["summary"]["p0_cicd_warning_step_runtime_enablement_gate_ready"] is True
assert (
data["summary"][
"p0_cicd_warning_step_runtime_enablement_active_blocker_count"
]
== 0
)
assert (
data["summary"][
"p0_cicd_warning_step_runtime_enablement_controlled_apply_allowed"
]
is True
)
assert (
data["summary"][
"p0_cicd_warning_step_runtime_enablement_workflow_dispatch_authorized"
]
is False
)
assert data["summary"]["p0_cicd_warning_step_runtime_enablement_switch"] == (
"AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED"
)
assert (
data["summary"][
"p0_cicd_warning_step_runtime_enablement_current_switch_default"
]
== "0"
)
assert (
data["summary"][
"p0_cicd_warning_step_runtime_enablement_desired_switch_value"
]
== "1"
)
assert (
data["summary"][
"p0_cicd_warning_step_runtime_enablement_trigger_performed"
]
is False
)
assert data["summary"]["production_deploy_status"] == "closure_verified"
assert data["summary"]["production_deploy_image_tag_matches_main"] is True
assert data["summary"]["backup_credential_escrow_intake_status"] == (

View File

@@ -16,6 +16,9 @@ from src.services.awoooi_gitea_onboarding_warning_step_owner_package import (
from src.services.awoooi_gitea_onboarding_warning_step_owner_response_preflight import (
load_latest_awoooi_gitea_onboarding_warning_step_owner_response_preflight,
)
from src.services.awoooi_gitea_onboarding_warning_step_runtime_enablement_gate import (
load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate,
)
from src.services.awoooi_gitea_onboarding_warning_step_template_copy_apply_gate import (
load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate,
)
@@ -345,6 +348,26 @@ def test_template_copy_receipt_supports_runtime_image_layout(tmp_path):
)
def test_warning_step_runtime_enablement_gate_loader_returns_controlled_package():
payload = load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate()
_assert_runtime_enablement_gate(payload)
def test_warning_step_runtime_enablement_gate_endpoint_returns_controlled_package():
app = FastAPI()
app.include_router(router, prefix="/api/v1")
client = TestClient(app)
response = client.get(
"/api/v1/agents/"
"awoooi-gitea-onboarding-warning-step-runtime-enablement-gate"
)
assert response.status_code == 200
_assert_runtime_enablement_gate(response.json())
def _assert_template_copy_receipt(payload: dict):
assert (
payload["schema_version"]
@@ -371,3 +394,47 @@ def _assert_template_copy_receipt(payload: dict):
assert payload["operation_boundaries"]["github_api_used"] is False
assert payload["operation_boundaries"]["secret_value_collection_allowed"] is False
assert payload["operation_boundaries"]["raw_session_or_sqlite_read_allowed"] is False
def _assert_runtime_enablement_gate(payload: dict):
assert (
payload["schema_version"]
== "awoooi_gitea_onboarding_warning_step_runtime_enablement_gate_v1"
)
assert payload["priority"] == "P0-004"
assert payload["scope"] == "gitea_onboarding_warning_step_runtime_enablement"
assert payload["status"] == "controlled_warning_step_runtime_enablement_gate_ready"
assert payload["active_blockers"] == []
assert payload["readback"]["source_receipt_status"] == (
"controlled_template_copy_receipt_ready"
)
assert payload["readback"]["runtime_enablement_switch"] == (
"AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED"
)
assert payload["readback"]["current_switch_default"] == "0"
assert payload["readback"]["desired_switch_value_after_controlled_apply"] == "1"
assert payload["controlled_apply"]["controlled_apply_allowed"] is True
assert payload["controlled_apply"]["workflow_runtime_enablement_authorized"] is True
assert payload["controlled_apply"]["workflow_dispatch_authorized"] is False
assert payload["controlled_apply"]["auto_push_or_pull_request_trigger_authorized"] is False
assert payload["controlled_apply"]["generic_runner_label_authorized"] is False
assert payload["target_selector"]["env_key"] == (
"AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED"
)
assert payload["target_selector"]["expected_current_value"] == "0"
assert payload["target_selector"]["desired_value_after_controlled_apply"] == "1"
assert payload["source_of_truth_diff"]["delta_kind"] == "env_switch_value_change"
assert payload["check_mode"]["enabled"] is True
assert payload["rollback"]["required"] is True
assert payload["post_apply_verifier"]["required"] is True
assert payload["rollups"]["template_copy_receipt_ready"] is True
assert payload["rollups"]["controlled_runtime_enablement_ready"] is True
assert payload["rollups"]["active_blocker_count"] == 0
assert payload["rollups"]["workflow_trigger_performed"] is False
assert payload["rollups"]["runtime_switch_changed"] is False
assert payload["operation_boundaries"]["readback_only"] is True
assert payload["operation_boundaries"]["controlled_runtime_enablement_gate_open"] is True
assert payload["operation_boundaries"]["workflow_file_modified_by_this_endpoint"] is False
assert payload["operation_boundaries"]["workflow_trigger_performed"] is False
assert payload["operation_boundaries"]["github_api_used"] is False
assert payload["operation_boundaries"]["secret_value_collection_allowed"] is False

View File

@@ -65,6 +65,7 @@ def test_p0_onboarding_readiness_sources_stay_on_controlled_runtime_profile() ->
"awoooi_gitea_onboarding_warning_step_template_copy_apply_gate.py",
"awoooi_gitea_onboarding_warning_step_template_copy_execution_plan.py",
"awoooi_gitea_onboarding_warning_step_template_copy_receipt.py",
"awoooi_gitea_onboarding_warning_step_runtime_enablement_gate.py",
"awoooi_new_product_onboarding_page_model.py",
"awoooi_onboarding_reminder_contract.py",
"awoooi_onboarding_source_contracts.py",