feat(api): expose onboarding template copy receipt

apps/api/src/services/awoooi_gitea_onboarding_warning_step_template_copy_receipt.py

@@ -0,0 +1,218 @@
+"""AWOOOI warning-step template copy receipt readback."""
+
+from __future__ import annotations
+
+import hashlib
+import re
+from pathlib import Path
+from typing import Any
+
+from src.services.awoooi_gitea_onboarding_warning_step_template_copy_apply_gate import (
+    load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate,
+)
+from src.services.snapshot_paths import resolve_repo_root
+
+_SCHEMA_VERSION = "awoooi_gitea_onboarding_warning_step_template_copy_receipt_v1"
+_TEMPLATE_RELATIVE_PATH = (
+    "docs/operations/templates/awoooi-gitea-onboarding-warning-step.workflow.yaml"
+)
+_WORKFLOW_RELATIVE_PATH = ".gitea/workflows/awoooi-onboarding-warning-step.yaml"
+_AUTO_BRANCH_EVENTS = ("push", "pull_request", "pull_request_target")
+_GENERIC_LABEL_PATTERNS = (
+    re.compile(r"^\s*runs-on:\s*.*\bubuntu-[A-Za-z0-9_.-]+\b", re.MULTILINE),
+    re.compile(r"^\s*runs-on:\s*.*\bself-hosted\b", re.MULTILINE),
+)
+
+
+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."""
+    root = repo_root or resolve_repo_root(Path(__file__))
+    template_path = root / _TEMPLATE_RELATIVE_PATH
+    workflow_path = root / _WORKFLOW_RELATIVE_PATH
+    template_text = template_path.read_text(encoding="utf-8") if template_path.exists() else ""
+    workflow_text = workflow_path.read_text(encoding="utf-8") if workflow_path.exists() else ""
+    apply_gate = load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate()
+    gate_readback = _dict(apply_gate.get("readback"))
+
+    active_blockers = _active_blockers(
+        template_path=template_path,
+        workflow_path=workflow_path,
+        template_text=template_text,
+        workflow_text=workflow_text,
+        gate_readback=gate_readback,
+    )
+    workflow_sha = _short_content_sha(workflow_text) if workflow_text else ""
+
+    return {
+        "schema_version": _SCHEMA_VERSION,
+        "priority": "P0-004",
+        "scope": "gitea_onboarding_warning_step_template_copy",
+        "status": (
+            "controlled_template_copy_receipt_ready"
+            if not active_blockers
+            else "blocked_template_copy_receipt_invalid"
+        ),
+        "readback": {
+            "workplan_id": "P0-004-TEMPLATE-COPY-CONTROLLED-APPLY",
+            "source_apply_gate_status": apply_gate.get("status"),
+            "template_copy_performed": workflow_path.is_file()
+            and template_text == workflow_text,
+            "source_template_path": _TEMPLATE_RELATIVE_PATH,
+            "destination_workflow_path": _WORKFLOW_RELATIVE_PATH,
+            "workflow_content_sha256_12": workflow_sha,
+            "safe_next_step": (
+                "open_next_gate_for_warning_step_runtime_enablement_after_pressure_guard"
+            ),
+        },
+        "target_selector": {
+            "selector_type": "controlled_workflow_template_copy",
+            "source_template_path": _TEMPLATE_RELATIVE_PATH,
+            "destination_workflow_path": _WORKFLOW_RELATIVE_PATH,
+            "active_workflow_file_created": True,
+        },
+        "source_of_truth_diff": {
+            "source": "controlled_apply_gate_readback",
+            "apply_gate_endpoint": (
+                "/api/v1/agents/"
+                "awoooi-gitea-onboarding-warning-step-template-copy-apply-gate"
+            ),
+            "receipt_endpoint": (
+                "/api/v1/agents/"
+                "awoooi-gitea-onboarding-warning-step-template-copy-receipt"
+            ),
+            "source_template_path": _TEMPLATE_RELATIVE_PATH,
+            "destination_workflow_path": _WORKFLOW_RELATIVE_PATH,
+        },
+        "check_mode": {
+            "enabled": True,
+            "required": True,
+            "checks": [
+                "template_file_exists",
+                "workflow_file_exists",
+                "workflow_matches_source_template",
+                "workflow_has_no_auto_branch_event",
+                "workflow_has_no_generic_runner_label",
+                "workflow_runtime_execution_switch_defaults_off",
+                "runner_pressure_guard_required",
+                "git_diff_check_required",
+            ],
+        },
+        "rollback": {
+            "required": True,
+            "strategy": "remove_copied_workflow_and_receipt_before_commit",
+            "paths": [
+                _WORKFLOW_RELATIVE_PATH,
+                "apps/api/src/services/"
+                "awoooi_gitea_onboarding_warning_step_template_copy_receipt.py",
+            ],
+        },
+        "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",
+            ],
+        },
+        "rollups": {
+            "template_file_present": template_path.is_file(),
+            "workflow_file_present": workflow_path.is_file(),
+            "workflow_matches_template": template_text == workflow_text,
+            "workflow_dispatch_declared": "workflow_dispatch:" in workflow_text,
+            "auto_branch_event_count": len(_auto_branch_event_hits(workflow_text)),
+            "generic_runner_label_count": len(_generic_label_hits(workflow_text)),
+            "fail_closed_execution_switch_present": _fail_closed_switch_present(
+                workflow_text
+            ),
+            "apply_gate_ready": _gate_ready(gate_readback),
+            "active_blocker_count": len(active_blockers),
+            "active_workflow_file_created": workflow_path.is_file(),
+            "workflow_trigger_performed": False,
+            "runner_pressure_guard_required": True,
+        },
+        "active_blockers": active_blockers,
+        "operation_boundaries": {
+            "controlled_template_copy_only": True,
+            "workflow_modification_allowed_by_gate": True,
+            "active_workflow_file_created": workflow_path.is_file(),
+            "workflow_dispatch_declared": "workflow_dispatch:" in workflow_text,
+            "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(
+    *,
+    template_path: Path,
+    workflow_path: Path,
+    template_text: str,
+    workflow_text: str,
+    gate_readback: dict[str, Any],
+) -> list[str]:
+    blockers: list[str] = []
+    if not _gate_ready(gate_readback):
+        blockers.append("template_copy_apply_gate_not_ready")
+    if not template_path.is_file():
+        blockers.append("source_template_file_missing")
+    if not workflow_path.is_file():
+        blockers.append("destination_workflow_file_missing")
+    if template_text != workflow_text:
+        blockers.append("destination_workflow_differs_from_source_template")
+    if _auto_branch_event_hits(workflow_text):
+        blockers.append("auto_branch_event_present_in_workflow")
+    if _generic_label_hits(workflow_text):
+        blockers.append("generic_runner_label_present_in_workflow")
+    if not _fail_closed_switch_present(workflow_text):
+        blockers.append("fail_closed_execution_switch_missing")
+    return blockers
+
+
+def _gate_ready(gate_readback: dict[str, Any]) -> bool:
+    return (
+        gate_readback.get("apply_allowed") is True
+        and gate_readback.get("controlled_apply_allowed") is True
+        and gate_readback.get("workflow_template_copy_authorized") is True
+        and gate_readback.get("workflow_trigger_authorized") is False
+        and gate_readback.get("auto_push_or_pull_request_trigger_authorized") is False
+        and gate_readback.get("generic_runner_label_authorized") is False
+        and gate_readback.get("runner_pressure_guard_required") is True
+    )
+
+
+def _auto_branch_event_hits(template_text: str) -> list[str]:
+    return [
+        event
+        for event in _AUTO_BRANCH_EVENTS
+        if re.search(rf"^\s*{event}\s*:", template_text, re.MULTILINE)
+        or re.search(rf"\bon:\s*\[[^\]]*\b{event}\b", template_text)
+    ]
+
+
+def _generic_label_hits(template_text: str) -> list[str]:
+    hits: list[str] = []
+    for pattern in _GENERIC_LABEL_PATTERNS:
+        hits.extend(pattern.findall(template_text))
+    return hits
+
+
+def _short_content_sha(template_text: str) -> str:
+    return hashlib.sha256(template_text.encode("utf-8")).hexdigest()[:12]
+
+
+def _fail_closed_switch_present(template_text: str) -> bool:
+    return (
+        'AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED: "0"' in template_text
+        and "AWOOOI_ONBOARDING_WARNING_STEP_EXECUTION_ENABLED == '1'" in template_text
+    )
+
+
+def _dict(value: Any) -> dict[str, Any]:
+    return value if isinstance(value, dict) else {}

apps/api/src/services/delivery_closure_workbench.py

@@ -12,6 +12,9 @@ from typing import Any
 from src.services.awoooi_gitea_onboarding_warning_step_template_copy_apply_gate import (
     load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate,
 )
+from src.services.awoooi_gitea_onboarding_warning_step_template_copy_receipt import (
+    load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt,
+)
 from src.services.awoooi_production_deploy_readback_blocker import (
     load_latest_awoooi_production_deploy_readback_blocker,
 )
@@ -52,6 +55,9 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
     cicd_template_copy_apply_gate = (
         load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate()
     )
+    cicd_template_copy_receipt = (
+        load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt()
+    )
     gitea = load_latest_gitea_workflow_runner_health()
     runtime = load_latest_runtime_surface_inventory()
     backup = load_latest_backup_dr_readiness_matrix()
@@ -63,6 +69,7 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
         private_inventory=private_inventory,
         cicd_baseline=cicd_baseline,
         cicd_template_copy_apply_gate=cicd_template_copy_apply_gate,
+        cicd_template_copy_receipt=cicd_template_copy_receipt,
         gitea=gitea,
         runtime=runtime,
         backup=backup,
@@ -78,6 +85,7 @@ def build_delivery_closure_workbench(
     private_inventory: dict[str, Any],
     cicd_baseline: dict[str, Any],
     cicd_template_copy_apply_gate: dict[str, Any],
+    cicd_template_copy_receipt: dict[str, Any],
     gitea: dict[str, Any],
     runtime: dict[str, Any],
     backup: dict[str, Any],
@@ -104,6 +112,15 @@ def build_delivery_closure_workbench(
     cicd_apply_gate_boundaries = _dict(
         cicd_template_copy_apply_gate.get("operation_boundaries")
     )
+    cicd_template_copy_receipt_readback = _dict(
+        cicd_template_copy_receipt.get("readback")
+    )
+    cicd_template_copy_receipt_rollups = _dict(
+        cicd_template_copy_receipt.get("rollups")
+    )
+    cicd_template_copy_receipt_boundaries = _dict(
+        cicd_template_copy_receipt.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"))
@@ -798,6 +815,64 @@ def build_delivery_closure_workbench(
                     cicd_apply_gate_readback.get("runner_pressure_guard_required")
                     is True
                 ),
+                "template_copy_receipt_status": str(
+                    cicd_template_copy_receipt.get("status") or ""
+                ),
+                "template_copy_receipt_ready": (
+                    cicd_template_copy_receipt.get("status")
+                    == "controlled_template_copy_receipt_ready"
+                ),
+                "template_copy_receipt_active_blocker_count": _int(
+                    cicd_template_copy_receipt_rollups.get("active_blocker_count")
+                ),
+                "source_template_path": str(
+                    cicd_template_copy_receipt_readback.get("source_template_path")
+                    or ""
+                ),
+                "destination_workflow_path": str(
+                    cicd_template_copy_receipt_readback.get(
+                        "destination_workflow_path"
+                    )
+                    or ""
+                ),
+                "source_template_file_present": (
+                    cicd_template_copy_receipt_rollups.get("template_file_present")
+                    is True
+                ),
+                "destination_workflow_file_present": (
+                    cicd_template_copy_receipt_rollups.get("workflow_file_present")
+                    is True
+                ),
+                "destination_workflow_matches_template": (
+                    cicd_template_copy_receipt_rollups.get("workflow_matches_template")
+                    is True
+                ),
+                "destination_workflow_dispatch_declared": (
+                    cicd_template_copy_receipt_rollups.get(
+                        "workflow_dispatch_declared"
+                    )
+                    is True
+                ),
+                "destination_workflow_auto_branch_event_count": _int(
+                    cicd_template_copy_receipt_rollups.get("auto_branch_event_count")
+                ),
+                "destination_workflow_generic_runner_label_count": _int(
+                    cicd_template_copy_receipt_rollups.get(
+                        "generic_runner_label_count"
+                    )
+                ),
+                "destination_workflow_active_file_created": (
+                    cicd_template_copy_receipt_boundaries.get(
+                        "active_workflow_file_created"
+                    )
+                    is True
+                ),
+                "destination_workflow_trigger_performed": (
+                    cicd_template_copy_receipt_boundaries.get(
+                        "workflow_trigger_performed"
+                    )
+                    is True
+                ),
                 "required_source_count": _int(
                     cicd_baseline_rollups.get("required_source_count")
                 ),
@@ -1191,6 +1266,53 @@ def build_delivery_closure_workbench(
                 cicd_apply_gate_readback.get("runner_pressure_guard_required")
                 is True
             ),
+            "p0_cicd_template_copy_receipt_status": str(
+                cicd_template_copy_receipt.get("status") or ""
+            ),
+            "p0_cicd_template_copy_receipt_ready": (
+                cicd_template_copy_receipt.get("status")
+                == "controlled_template_copy_receipt_ready"
+            ),
+            "p0_cicd_template_copy_receipt_active_blocker_count": _int(
+                cicd_template_copy_receipt_rollups.get("active_blocker_count")
+            ),
+            "p0_cicd_template_copy_source_template_path": str(
+                cicd_template_copy_receipt_readback.get("source_template_path")
+                or ""
+            ),
+            "p0_cicd_template_copy_destination_workflow_path": str(
+                cicd_template_copy_receipt_readback.get("destination_workflow_path")
+                or ""
+            ),
+            "p0_cicd_template_copy_source_template_file_present": (
+                cicd_template_copy_receipt_rollups.get("template_file_present")
+                is True
+            ),
+            "p0_cicd_template_copy_destination_workflow_file_present": (
+                cicd_template_copy_receipt_rollups.get("workflow_file_present") is True
+            ),
+            "p0_cicd_template_copy_destination_workflow_matches_template": (
+                cicd_template_copy_receipt_rollups.get("workflow_matches_template")
+                is True
+            ),
+            "p0_cicd_template_copy_workflow_dispatch_declared": (
+                cicd_template_copy_receipt_rollups.get("workflow_dispatch_declared")
+                is True
+            ),
+            "p0_cicd_template_copy_auto_branch_event_count": _int(
+                cicd_template_copy_receipt_rollups.get("auto_branch_event_count")
+            ),
+            "p0_cicd_template_copy_generic_runner_label_count": _int(
+                cicd_template_copy_receipt_rollups.get("generic_runner_label_count")
+            ),
+            "p0_cicd_template_copy_active_workflow_file_created": (
+                cicd_template_copy_receipt_boundaries.get("active_workflow_file_created")
+                is True
+            ),
+            "p0_cicd_template_copy_workflow_trigger_performed": (
+                cicd_template_copy_receipt_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"