feat(ci): enable warning step runtime switch
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 23s
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 01:26:10 +08:00
parent 93ae219a12
commit d1e8400ae9
9 changed files with 123 additions and 49 deletions

View File

@@ -22,6 +22,8 @@ def load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate() -
receipt_rollups = _dict(receipt.get("rollups"))
receipt_boundaries = _dict(receipt.get("operation_boundaries"))
receipt_readback = _dict(receipt.get("readback"))
switch_value = str(receipt_rollups.get("runtime_execution_switch_value") or "0")
switch_enabled = switch_value == "1"
receipt_ready = (
receipt.get("status") == _RECEIPT_READY_STATUS
and receipt.get("active_blockers") == []
@@ -30,7 +32,7 @@ def load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate() -
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_rollups.get("runtime_execution_switch_controlled") is True
and receipt_boundaries.get("workflow_trigger_performed") is False
)
active_blockers = _active_blockers(
@@ -44,7 +46,9 @@ def load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate() -
"priority": "P0-004",
"scope": "gitea_onboarding_warning_step_runtime_enablement",
"status": (
"controlled_warning_step_runtime_enablement_gate_ready"
"controlled_warning_step_runtime_enabled"
if switch_enabled and not active_blockers
else "controlled_warning_step_runtime_enablement_gate_ready"
if not active_blockers
else "blocked_warning_step_runtime_enablement_gate"
),
@@ -64,16 +68,19 @@ def load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate() -
"workflow_content_sha256_12"
),
"runtime_enablement_switch": "AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED",
"current_switch_default": "0",
"current_switch_default": switch_value,
"desired_switch_value_after_controlled_apply": "1",
"safe_next_step": (
"prepare_controlled_switch_flip_with_pressure_guard_then_verify_no_auto_branch_trigger"
"runtime_enablement_switch_enabled_wait_for_manual_dispatch_evidence"
if switch_enabled
else "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,
"controlled_apply_allowed": not active_blockers and not switch_enabled,
"workflow_runtime_enablement_authorized": not active_blockers
and not switch_enabled,
"workflow_dispatch_authorized": False,
"auto_push_or_pull_request_trigger_authorized": False,
"generic_runner_label_authorized": False,
@@ -88,12 +95,16 @@ def load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate() -
"selector_type": "workflow_runtime_enablement_switch",
"workflow_path": _WORKFLOW_PATH,
"env_key": "AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED",
"expected_current_value": "0",
"expected_current_value": switch_value,
"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",
"current_state": (
"warning_step_runtime_job_enabled_for_manual_dispatch_only"
if switch_enabled
else "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,
@@ -149,11 +160,16 @@ def load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate() -
"fail_closed_execution_switch_present": (
receipt_rollups.get("fail_closed_execution_switch_present") is True
),
"runtime_execution_switch_controlled": (
receipt_rollups.get("runtime_execution_switch_controlled") is True
),
"runtime_execution_switch_value": switch_value,
"runtime_execution_switch_enabled": switch_enabled,
"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,
"runtime_switch_changed": switch_enabled,
},
"active_blockers": active_blockers,
"operation_boundaries": {
@@ -184,8 +200,8 @@ def _active_blockers(
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_rollups.get("runtime_execution_switch_controlled") is not True:
blockers.append("runtime_execution_switch_not_controlled")
if receipt_boundaries.get("workflow_trigger_performed") is not False:
blockers.append("workflow_trigger_already_performed")
return _unique(blockers)

View File

@@ -17,7 +17,7 @@ _SCHEMA_VERSION = "awoooi_gitea_onboarding_warning_step_template_copy_receipt_v1
_RECEIPT_SCHEMA_VERSION = (
"awoooi_gitea_onboarding_warning_step_template_copy_receipt_snapshot_v1"
)
_EXPECTED_WORKFLOW_SHA256_12 = "70ecac9e4b59"
_EXPECTED_WORKFLOW_SHA256_12 = "23b59f29cb77"
_TEMPLATE_RELATIVE_PATH = (
"docs/operations/templates/awoooi-gitea-onboarding-warning-step.workflow.yaml"
)
@@ -36,7 +36,7 @@ _GENERIC_LABEL_PATTERNS = (
def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
repo_root: Path | None = None,
) -> dict[str, Any]:
"""Return the fail-closed warning-step workflow copy receipt."""
"""Return the warning-step workflow copy receipt."""
root = repo_root or resolve_repo_root(Path(__file__))
template_path = root / _TEMPLATE_RELATIVE_PATH
workflow_path = root / _WORKFLOW_RELATIVE_PATH
@@ -54,6 +54,9 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
effective_workflow_sha = (
_short_content_sha(effective_workflow_text) if effective_workflow_text else ""
)
runtime_switch_value = _runtime_execution_switch_value(
receipt, effective_workflow_text
)
template_copy_recorded = _template_copy_recorded(
receipt=receipt,
template_sha=template_sha,
@@ -137,7 +140,7 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
"workflow_matches_source_template_or_receipt_hash_matches",
"workflow_has_no_auto_branch_event",
"workflow_has_no_generic_runner_label",
"workflow_runtime_execution_switch_defaults_off",
"workflow_runtime_execution_switch_controlled",
"runner_pressure_guard_required",
"git_diff_check_required",
],
@@ -182,6 +185,12 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
effective_workflow_text
)
or receipt.get("fail_closed_execution_switch_present") is True,
"runtime_execution_switch_controlled": _runtime_execution_switch_controlled(
receipt, effective_workflow_text
),
"runtime_execution_switch_value": runtime_switch_value,
"runtime_execution_switch_enabled": runtime_switch_value == "1",
"runtime_enablement_switch_changed": runtime_switch_value == "1",
"apply_gate_ready": _gate_ready(gate_readback),
"active_blocker_count": len(active_blockers),
"active_workflow_file_created": template_copy_recorded,
@@ -233,11 +242,8 @@ def _active_blockers(
blockers.append("auto_branch_event_present_in_workflow")
if _generic_runner_label_count(receipt, effective_workflow_text) > 0:
blockers.append("generic_runner_label_present_in_workflow")
if not (
_fail_closed_switch_present(effective_workflow_text)
or receipt.get("fail_closed_execution_switch_present") is True
):
blockers.append("fail_closed_execution_switch_missing")
if not _runtime_execution_switch_controlled(receipt, effective_workflow_text):
blockers.append("runtime_execution_switch_not_controlled")
return blockers
@@ -357,5 +363,26 @@ def _fail_closed_switch_present(template_text: str) -> bool:
)
def _runtime_execution_switch_value(receipt: dict[str, Any], template_text: str) -> str:
match = re.search(
r'AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED:\s*"([01])"',
template_text,
)
if match:
return match.group(1)
value = receipt.get("runtime_execution_switch_value")
return value if value in {"0", "1"} else ""
def _runtime_execution_switch_controlled(
receipt: dict[str, Any],
template_text: str,
) -> bool:
return (
_runtime_execution_switch_value(receipt, template_text) in {"0", "1"}
and "AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED == '1'" in template_text
) or receipt.get("runtime_execution_switch_controlled") is True
def _dict(value: Any) -> dict[str, Any]:
return value if isinstance(value, dict) else {}

View File

@@ -898,7 +898,10 @@ def build_delivery_closure_workbench(
),
"runtime_enablement_gate_ready": (
cicd_runtime_enablement_gate.get("status")
== "controlled_warning_step_runtime_enablement_gate_ready"
in {
"controlled_warning_step_runtime_enablement_gate_ready",
"controlled_warning_step_runtime_enabled",
}
),
"runtime_enablement_gate_active_blocker_count": _int(
cicd_runtime_enablement_rollups.get("active_blocker_count")
@@ -1380,7 +1383,10 @@ def build_delivery_closure_workbench(
),
"p0_cicd_warning_step_runtime_enablement_gate_ready": (
cicd_runtime_enablement_gate.get("status")
== "controlled_warning_step_runtime_enablement_gate_ready"
in {
"controlled_warning_step_runtime_enablement_gate_ready",
"controlled_warning_step_runtime_enabled",
}
),
"p0_cicd_warning_step_runtime_enablement_active_blocker_count": _int(
cicd_runtime_enablement_rollups.get("active_blocker_count")

View File

@@ -372,7 +372,7 @@ def _assert_delivery_workbench_shape(data: dict):
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"
"controlled_warning_step_runtime_enabled"
)
assert data["summary"]["p0_cicd_warning_step_runtime_enablement_gate_ready"] is True
assert (
@@ -385,7 +385,7 @@ def _assert_delivery_workbench_shape(data: dict):
data["summary"][
"p0_cicd_warning_step_runtime_enablement_controlled_apply_allowed"
]
is True
is False
)
assert (
data["summary"][
@@ -400,7 +400,7 @@ def _assert_delivery_workbench_shape(data: dict):
data["summary"][
"p0_cicd_warning_step_runtime_enablement_current_switch_default"
]
== "0"
== "1"
)
assert (
data["summary"][

View File

@@ -286,7 +286,7 @@ def test_warning_step_template_copy_is_fail_closed_and_pressure_guarded():
assert "workflow-shape:" in workflow
assert "Confirm fail-closed warning-step workflow shape" in workflow
assert "runs-on: awoooi-non110-host" in workflow
assert 'AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED: "0"' in workflow
assert 'AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED: "1"' in workflow
assert "ops/runner/guard-gitea-runner-pressure.py --root ." in workflow
@@ -377,7 +377,7 @@ def _assert_template_copy_receipt(payload: dict):
assert payload["status"] == "controlled_template_copy_receipt_ready"
assert payload["active_blockers"] == []
assert payload["readback"]["template_copy_performed"] is True
assert payload["readback"]["workflow_content_sha256_12"] == "70ecac9e4b59"
assert payload["readback"]["workflow_content_sha256_12"] == "23b59f29cb77"
assert payload["target_selector"]["active_workflow_file_created"] is True
assert payload["rollups"]["template_file_present"] is True
assert payload["rollups"]["receipt_snapshot_present"] is True
@@ -385,7 +385,12 @@ def _assert_template_copy_receipt(payload: dict):
assert payload["rollups"]["workflow_matches_template"] is True
assert payload["rollups"]["auto_branch_event_count"] == 0
assert payload["rollups"]["generic_runner_label_count"] == 0
assert payload["rollups"]["expected_workflow_sha256_12"] == "70ecac9e4b59"
assert payload["rollups"]["expected_workflow_sha256_12"] == "23b59f29cb77"
assert payload["rollups"]["fail_closed_execution_switch_present"] is False
assert payload["rollups"]["runtime_execution_switch_controlled"] is True
assert payload["rollups"]["runtime_execution_switch_value"] == "1"
assert payload["rollups"]["runtime_execution_switch_enabled"] is True
assert payload["rollups"]["runtime_enablement_switch_changed"] is True
assert payload["rollups"]["apply_gate_ready"] is True
assert payload["rollups"]["workflow_trigger_performed"] is False
assert payload["operation_boundaries"]["controlled_template_copy_only"] is True
@@ -403,7 +408,7 @@ def _assert_runtime_enablement_gate(payload: dict):
)
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["status"] == "controlled_warning_step_runtime_enabled"
assert payload["active_blockers"] == []
assert payload["readback"]["source_receipt_status"] == (
"controlled_template_copy_receipt_ready"
@@ -411,17 +416,17 @@ def _assert_runtime_enablement_gate(payload: dict):
assert payload["readback"]["runtime_enablement_switch"] == (
"AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED"
)
assert payload["readback"]["current_switch_default"] == "0"
assert payload["readback"]["current_switch_default"] == "1"
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"]["controlled_apply_allowed"] is False
assert payload["controlled_apply"]["workflow_runtime_enablement_authorized"] is False
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"]["expected_current_value"] == "1"
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
@@ -431,7 +436,10 @@ def _assert_runtime_enablement_gate(payload: dict):
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["rollups"]["runtime_execution_switch_controlled"] is True
assert payload["rollups"]["runtime_execution_switch_value"] == "1"
assert payload["rollups"]["runtime_execution_switch_enabled"] is True
assert payload["rollups"]["runtime_switch_changed"] is True
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