fix(api): support runtime template copy receipt

apps/api/src/services/awoooi_gitea_onboarding_warning_step_template_copy_receipt.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import hashlib
+import json
 import re
 from pathlib import Path
 from typing import Any
@@ -13,11 +14,18 @@ from src.services.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"
+_RECEIPT_SCHEMA_VERSION = (
+    "awoooi_gitea_onboarding_warning_step_template_copy_receipt_snapshot_v1"
+)
+_EXPECTED_WORKFLOW_SHA256_12 = "70ecac9e4b59"
 _TEMPLATE_RELATIVE_PATH = (
     "docs/operations/templates/awoooi-gitea-onboarding-warning-step.workflow.yaml"
 )
 _WORKFLOW_RELATIVE_PATH = ".gitea/workflows/awoooi-onboarding-warning-step.yaml"
-_EXPECTED_WORKFLOW_SHA256_12 = "70ecac9e4b59"
+_RECEIPT_RELATIVE_PATH = (
+    "docs/operations/awoooi-gitea-onboarding-warning-step-template-copy-receipt."
+    "snapshot.json"
+)
 _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),
@@ -32,36 +40,45 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
     root = repo_root or resolve_repo_root(Path(__file__))
     template_path = root / _TEMPLATE_RELATIVE_PATH
     workflow_path = root / _WORKFLOW_RELATIVE_PATH
+    receipt_path = root / _RECEIPT_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 ""
+    receipt = _load_receipt_snapshot(receipt_path)
     effective_workflow_text = workflow_text or template_text
     workflow_content_source = (
-        "repo_workflow_file"
-        if workflow_text
-        else "packaged_template_digest_fallback"
+        "repo_workflow_file" if workflow_text else "packaged_template_digest_fallback"
+    )
+    template_sha = _short_content_sha(template_text) if template_text else ""
+    workflow_sha = _short_content_sha(workflow_text) if workflow_text else ""
+    effective_workflow_sha = (
+        _short_content_sha(effective_workflow_text) if effective_workflow_text else ""
+    )
+    template_copy_recorded = _template_copy_recorded(
+        receipt=receipt,
+        template_sha=template_sha,
+        workflow_sha=workflow_sha,
+        workflow_visible=workflow_path.is_file(),
+        template_text=template_text,
+        workflow_text=workflow_text,
+    )
+    workflow_matches_template = _workflow_matches_template(
+        receipt=receipt,
+        template_sha=template_sha,
+        workflow_sha=workflow_sha,
+        workflow_visible=workflow_path.is_file(),
+        template_text=template_text,
+        workflow_text=workflow_text,
     )
     apply_gate = load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate()
     gate_readback = _dict(apply_gate.get("readback"))
-    workflow_sha = (
-        _short_content_sha(effective_workflow_text) if effective_workflow_text else ""
-    )
-    committed_workflow_file_created = (
-        template_path.is_file() and workflow_sha == _EXPECTED_WORKFLOW_SHA256_12
-    )
-    workflow_matches_template = (
-        bool(effective_workflow_text)
-        and template_text == effective_workflow_text
-        and workflow_sha == _EXPECTED_WORKFLOW_SHA256_12
-    )
 
     active_blockers = _active_blockers(
         template_path=template_path,
-        workflow_path=workflow_path,
-        template_text=template_text,
-        workflow_text=workflow_text,
-        effective_workflow_text=effective_workflow_text,
-        committed_workflow_file_created=committed_workflow_file_created,
+        receipt=receipt,
+        template_copy_recorded=template_copy_recorded,
         workflow_matches_template=workflow_matches_template,
+        effective_workflow_text=effective_workflow_text,
         gate_readback=gate_readback,
     )
 
@@ -77,11 +94,15 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
         "readback": {
             "workplan_id": "P0-004-TEMPLATE-COPY-CONTROLLED-APPLY",
             "source_apply_gate_status": apply_gate.get("status"),
-            "template_copy_performed": committed_workflow_file_created
-            and workflow_matches_template,
+            "template_copy_performed": template_copy_recorded,
+            "receipt_snapshot_path": _RECEIPT_RELATIVE_PATH,
             "source_template_path": _TEMPLATE_RELATIVE_PATH,
             "destination_workflow_path": _WORKFLOW_RELATIVE_PATH,
-            "workflow_content_sha256_12": workflow_sha,
+            "source_template_content_sha256_12": template_sha,
+            "destination_workflow_content_sha256_12": _destination_workflow_sha(
+                receipt, workflow_sha
+            ),
+            "workflow_content_sha256_12": effective_workflow_sha,
             "workflow_content_source": workflow_content_source,
             "safe_next_step": (
                 "open_next_gate_for_warning_step_runtime_enablement_after_pressure_guard"
@@ -111,8 +132,9 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
             "required": True,
             "checks": [
                 "template_file_exists",
-                "workflow_file_exists",
-                "workflow_matches_source_template",
+                "receipt_snapshot_exists",
+                "workflow_file_exists_or_receipt_records_copy",
+                "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",
@@ -125,6 +147,7 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
             "strategy": "remove_copied_workflow_and_receipt_before_commit",
             "paths": [
                 _WORKFLOW_RELATIVE_PATH,
+                _RECEIPT_RELATIVE_PATH,
                 "apps/api/src/services/"
                 "awoooi_gitea_onboarding_warning_step_template_copy_receipt.py",
             ],
@@ -139,24 +162,29 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
         },
         "rollups": {
             "template_file_present": template_path.is_file(),
-            "workflow_file_present": committed_workflow_file_created,
+            "receipt_snapshot_present": bool(receipt),
+            "workflow_file_present": template_copy_recorded,
             "runtime_image_workflow_file_present": workflow_path.is_file(),
+            "workflow_file_visible_in_runtime_image": workflow_path.is_file(),
             "workflow_matches_template": workflow_matches_template,
             "workflow_content_source": workflow_content_source,
             "expected_workflow_sha256_12": _EXPECTED_WORKFLOW_SHA256_12,
-            "workflow_dispatch_declared": "workflow_dispatch:" in effective_workflow_text,
-            "auto_branch_event_count": len(
-                _auto_branch_event_hits(effective_workflow_text)
+            "workflow_dispatch_declared": _workflow_dispatch_declared(
+                receipt, effective_workflow_text
             ),
-            "generic_runner_label_count": len(
-                _generic_label_hits(effective_workflow_text)
+            "auto_branch_event_count": _auto_branch_event_count(
+                receipt, effective_workflow_text
+            ),
+            "generic_runner_label_count": _generic_runner_label_count(
+                receipt, effective_workflow_text
             ),
             "fail_closed_execution_switch_present": _fail_closed_switch_present(
                 effective_workflow_text
-            ),
+            )
+            or receipt.get("fail_closed_execution_switch_present") is True,
             "apply_gate_ready": _gate_ready(gate_readback),
             "active_blocker_count": len(active_blockers),
-            "active_workflow_file_created": committed_workflow_file_created,
+            "active_workflow_file_created": template_copy_recorded,
             "workflow_trigger_performed": False,
             "runner_pressure_guard_required": True,
         },
@@ -164,9 +192,12 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
         "operation_boundaries": {
             "controlled_template_copy_only": True,
             "workflow_modification_allowed_by_gate": True,
-            "active_workflow_file_created": committed_workflow_file_created,
+            "active_workflow_file_created": template_copy_recorded,
             "runtime_image_workflow_file_present": workflow_path.is_file(),
-            "workflow_dispatch_declared": "workflow_dispatch:" in effective_workflow_text,
+            "active_workflow_file_visible_in_runtime_image": workflow_path.is_file(),
+            "workflow_dispatch_declared": _workflow_dispatch_declared(
+                receipt, effective_workflow_text
+            ),
             "workflow_trigger_performed": False,
             "auto_push_or_pull_request_trigger_allowed": False,
             "generic_runner_label_allowed": False,
@@ -181,12 +212,10 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
 def _active_blockers(
     *,
     template_path: Path,
-    workflow_path: Path,
-    template_text: str,
-    workflow_text: str,
-    effective_workflow_text: str,
-    committed_workflow_file_created: bool,
+    receipt: dict[str, Any],
+    template_copy_recorded: bool,
     workflow_matches_template: bool,
+    effective_workflow_text: str,
     gate_readback: dict[str, Any],
 ) -> list[str]:
     blockers: list[str] = []
@@ -194,17 +223,20 @@ def _active_blockers(
         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() and not committed_workflow_file_created:
-        blockers.append("destination_workflow_file_missing")
-    if workflow_path.is_file() and template_text != workflow_text:
-        blockers.append("destination_workflow_differs_from_source_template")
+    if not receipt:
+        blockers.append("template_copy_receipt_snapshot_missing")
+    if not template_copy_recorded:
+        blockers.append("template_copy_not_recorded")
     if not workflow_matches_template:
-        blockers.append("destination_workflow_digest_not_confirmed")
-    if _auto_branch_event_hits(effective_workflow_text):
+        blockers.append("destination_workflow_differs_from_source_template")
+    if _auto_branch_event_count(receipt, effective_workflow_text) > 0:
         blockers.append("auto_branch_event_present_in_workflow")
-    if _generic_label_hits(effective_workflow_text):
+    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):
+    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")
     return blockers
 
@@ -221,6 +253,69 @@ def _gate_ready(gate_readback: dict[str, Any]) -> bool:
     )
 
 
+def _load_receipt_snapshot(path: Path) -> dict[str, Any]:
+    if not path.exists():
+        return {}
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path}: receipt snapshot must be an object")
+    if payload.get("schema_version") != _RECEIPT_SCHEMA_VERSION:
+        raise ValueError(f"{path}: unexpected receipt snapshot schema")
+    return payload
+
+
+def _template_copy_recorded(
+    *,
+    receipt: dict[str, Any],
+    template_sha: str,
+    workflow_sha: str,
+    workflow_visible: bool,
+    template_text: str,
+    workflow_text: str,
+) -> bool:
+    if workflow_visible:
+        return bool(template_text) and template_text == workflow_text
+    return (
+        receipt.get("template_copy_recorded") is True
+        and str(receipt.get("source_template_content_sha256_12") or "") == template_sha
+        and str(receipt.get("destination_workflow_content_sha256_12") or "")
+        == template_sha
+        and receipt.get("runtime_image_excludes_dot_gitea") is True
+        and not workflow_sha
+    )
+
+
+def _workflow_matches_template(
+    *,
+    receipt: dict[str, Any],
+    template_sha: str,
+    workflow_sha: str,
+    workflow_visible: bool,
+    template_text: str,
+    workflow_text: str,
+) -> bool:
+    if workflow_visible:
+        return bool(template_text) and template_text == workflow_text
+    return (
+        bool(template_sha)
+        and str(receipt.get("source_template_content_sha256_12") or "") == template_sha
+        and str(receipt.get("destination_workflow_content_sha256_12") or "")
+        == template_sha
+        and not workflow_sha
+    )
+
+
+def _destination_workflow_sha(receipt: dict[str, Any], workflow_sha: str) -> str:
+    return workflow_sha or str(receipt.get("destination_workflow_content_sha256_12") or "")
+
+
+def _workflow_dispatch_declared(receipt: dict[str, Any], workflow_text: str) -> bool:
+    return "workflow_dispatch:" in workflow_text or (
+        receipt.get("workflow_dispatch_declared") is True
+    )
+
+
 def _auto_branch_event_hits(template_text: str) -> list[str]:
     return [
         event
@@ -230,6 +325,13 @@ def _auto_branch_event_hits(template_text: str) -> list[str]:
     ]
 
 
+def _auto_branch_event_count(receipt: dict[str, Any], workflow_text: str) -> int:
+    if workflow_text:
+        return len(_auto_branch_event_hits(workflow_text))
+    value = receipt.get("auto_branch_event_count")
+    return value if isinstance(value, int) else 0
+
+
 def _generic_label_hits(template_text: str) -> list[str]:
     hits: list[str] = []
     for pattern in _GENERIC_LABEL_PATTERNS:
@@ -237,6 +339,13 @@ def _generic_label_hits(template_text: str) -> list[str]:
     return hits
 
 
+def _generic_runner_label_count(receipt: dict[str, Any], workflow_text: str) -> int:
+    if workflow_text:
+        return len(_generic_label_hits(workflow_text))
+    value = receipt.get("generic_runner_label_count")
+    return value if isinstance(value, int) else 0
+
+
 def _short_content_sha(template_text: str) -> str:
     return hashlib.sha256(template_text.encode("utf-8")).hexdigest()[:12]