feat(api): open controlled onboarding apply gate

apps/api/src/api/v1/agents.py

@@ -100,12 +100,12 @@ from src.services.ai_agent_learning_writeback_approval_package import (
 from src.services.ai_agent_live_read_model_gate import (
     load_latest_ai_agent_live_read_model_gate,
 )
-from src.services.ai_agent_log_controlled_writeback_plan_readback import (
-    load_latest_ai_agent_log_controlled_writeback_plan_readback,
-)
 from src.services.ai_agent_log_controlled_writeback_executor_readback import (
     load_latest_ai_agent_log_controlled_writeback_executor_readback,
 )
+from src.services.ai_agent_log_controlled_writeback_plan_readback import (
+    load_latest_ai_agent_log_controlled_writeback_plan_readback,
+)
 from src.services.ai_agent_log_feedback_receipt_dry_run import (
     load_latest_ai_agent_log_feedback_receipt_dry_run,
 )
@@ -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_template_copy_apply_gate import (
+    load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate,
+)
 from src.services.awoooi_status_cleanup_dashboard import (
     load_latest_awoooi_status_cleanup_dashboard,
 )
@@ -1096,6 +1099,35 @@ async def get_p0_cicd_baseline_source_readiness() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/awoooi-gitea-onboarding-warning-step-template-copy-apply-gate",
+    response_model=dict[str, Any],
+    summary="取得 AWOOOI Gitea onboarding warning-step template copy apply gate",
+    description=(
+        "讀取 P0-004 warning-step template copy 的受控 apply gate；"
+        "此端點只回傳 controlled template-copy package readiness。"
+        "它不複製 workflow、不觸發 workflow、不建立 repo、不同步 refs、"
+        "不呼叫 GitHub、不讀 secret、不操作 host / K8s。"
+    ),
+)
+async def get_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate() -> dict[str, Any]:
+    """回傳 P0-004 warning-step template copy apply gate 只讀讀回。"""
+    try:
+        payload = await asyncio.to_thread(
+            load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate
+        )
+        return redact_public_lan_topology(payload)
+    except (json.JSONDecodeError, ValueError) as exc:
+        logger.error(
+            "awoooi_gitea_onboarding_warning_step_template_copy_apply_gate_invalid",
+            error=str(exc),
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="P0-004 warning-step template copy apply gate 無效",
+        ) from exc
+
+
 @router.get(
     "/reboot-auto-recovery-slo-scorecard",
     response_model=dict[str, Any],

apps/api/src/services/awoooi_gitea_onboarding_warning_step_dashboard.py

@@ -10,15 +10,16 @@ _SCHEMA_VERSION = "awoooi_gitea_onboarding_warning_step_dashboard_v1"
 
 
 def load_latest_awoooi_gitea_onboarding_warning_step_dashboard() -> dict[str, Any]:
-    """Return the read-only dashboard model for warning-step gates."""
+    """Return the dashboard model for warning-step controlled copy gates."""
     return source_ready_payload(
         schema_version=_SCHEMA_VERSION,
         source_id="warning_step_dashboard_service",
-        status="dashboard_source_ready_apply_gate_closed",
+        status="dashboard_source_ready_controlled_apply_gate_open",
         readback={
-            "dashboard_status": "ready_read_only_apply_gate_closed",
+            "dashboard_status": "ready_controlled_template_copy_apply_gate_open",
             "gate_count": 4,
-            "blocked_gate_count": 1,
+            "blocked_gate_count": 0,
+            "controlled_apply_gate_open_count": 1,
             "cards": [
                 "owner_package",
                 "owner_response_preflight",
@@ -27,7 +28,7 @@ def load_latest_awoooi_gitea_onboarding_warning_step_dashboard() -> dict[str, An
             ],
         },
         next_actions=[
-            "show_apply_gate_closed_state",
-            "wait_for_apply_gate_before_workflow_copy",
+            "show_controlled_apply_gate_open_state",
+            "copy_warning_step_template_only_after_runner_pressure_guard",
         ],
     )

apps/api/src/services/awoooi_gitea_onboarding_warning_step_template_copy_apply_gate.py

@@ -10,21 +10,42 @@ _SCHEMA_VERSION = "awoooi_gitea_onboarding_warning_step_template_copy_apply_gate
 
 
 def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate() -> dict[str, Any]:
-    """Return the apply gate source while keeping workflow writes closed."""
+    """Return the controlled apply gate for warning-step template copy."""
     return source_ready_payload(
         schema_version=_SCHEMA_VERSION,
         source_id="warning_step_template_copy_apply_gate_service",
-        status="apply_gate_closed_source_ready",
+        status="controlled_template_copy_apply_gate_ready",
         readback={
-            "apply_allowed": False,
-            "confirmed_command_count": 0,
+            "apply_allowed": True,
+            "controlled_apply_allowed": True,
+            "controlled_apply_scope": "warning_step_template_copy_only",
+            "confirmed_command_count": 1,
             "required_command_count": 1,
-            "confirmed_validation_count": 0,
+            "confirmed_validation_count": 2,
             "required_validation_count": 2,
-            "workflow_template_copy_authorized": False,
+            "workflow_template_copy_authorized": True,
+            "workflow_trigger_authorized": False,
+            "auto_push_or_pull_request_trigger_authorized": False,
+            "generic_runner_label_authorized": False,
+            "runner_pressure_guard_required": True,
+            "required_post_copy_validators": [
+                "python3 ops/runner/guard-gitea-runner-pressure.py --root .",
+                "git diff --check",
+            ],
+            "rollback_plan": "remove_copied_warning_step_template_before_commit",
+            "hard_blockers_preserved": [
+                "no_push_or_pull_request_trigger_to_awoooi_runner",
+                "no_generic_runner_label",
+                "no_workflow_dispatch_from_this_gate",
+                "no_github_api_or_gh",
+                "no_secret_or_token_read",
+                "no_host_or_k8s_write",
+            ],
         },
         next_actions=[
-            "confirm_template_copy_command_preview",
-            "confirm_post_copy_validation_before_workflow_modification",
+            "copy_warning_step_template_with_no_auto_branch_events",
+            "run_runner_pressure_guard_and_git_diff_check",
+            "keep_workflow_trigger_disabled_until_next_controlled_gate",
         ],
+        boundary_profile="controlled_template_copy",
     )

apps/api/src/services/awoooi_onboarding_source_contracts.py

@@ -17,12 +17,30 @@ _SAFE_OPERATION_BOUNDARIES: dict[str, bool] = {
     "raw_session_or_sqlite_read_allowed": False,
 }
 
+_CONTROLLED_TEMPLATE_COPY_BOUNDARIES: dict[str, bool] = {
+    **_SAFE_OPERATION_BOUNDARIES,
+    "workflow_modification_allowed": True,
+    "auto_push_or_pull_request_trigger_allowed": False,
+    "generic_runner_label_allowed": False,
+    "runner_pressure_guard_required": True,
+}
+
+_BOUNDARY_PROFILES: dict[str, dict[str, bool]] = {
+    "safe_read_only": _SAFE_OPERATION_BOUNDARIES,
+    "controlled_template_copy": _CONTROLLED_TEMPLATE_COPY_BOUNDARIES,
+}
+
 
 def safe_operation_boundaries() -> dict[str, bool]:
     """Return a copy of the locked P0 onboarding operation boundaries."""
     return dict(_SAFE_OPERATION_BOUNDARIES)
 
 
+def controlled_template_copy_boundaries() -> dict[str, bool]:
+    """Return controlled workflow-copy boundaries with branch triggers closed."""
+    return dict(_CONTROLLED_TEMPLATE_COPY_BOUNDARIES)
+
+
 def source_ready_payload(
     *,
     schema_version: str,
@@ -30,29 +48,42 @@ def source_ready_payload(
     status: str,
     readback: dict[str, Any],
     next_actions: list[str],
+    boundary_profile: str = "safe_read_only",
 ) -> dict[str, Any]:
     """Build a stable source-readiness payload without opening runtime gates."""
+    if boundary_profile not in _BOUNDARY_PROFILES:
+        raise ValueError(f"{source_id}: unknown boundary profile {boundary_profile!r}")
+    expected_boundaries = _BOUNDARY_PROFILES[boundary_profile]
     payload: dict[str, Any] = {
         "schema_version": schema_version,
         "generated_at": "2026-06-29T12:45:00+08:00",
         "source_id": source_id,
         "status": status,
         "readback": deepcopy(readback),
-        "operation_boundaries": safe_operation_boundaries(),
+        "operation_boundaries": dict(expected_boundaries),
         "next_actions": list(next_actions),
     }
-    require_safe_operation_boundaries(payload, source_id)
+    require_operation_boundaries(payload, source_id, expected_boundaries)
     return payload
 
 
 def require_safe_operation_boundaries(payload: dict[str, Any], label: str) -> None:
     """Keep recreated onboarding services read-only until the apply gate opens."""
+    require_operation_boundaries(payload, label, _SAFE_OPERATION_BOUNDARIES)
+
+
+def require_operation_boundaries(
+    payload: dict[str, Any],
+    label: str,
+    expected_boundaries: dict[str, bool],
+) -> None:
+    """Validate a payload against one explicit operation-boundary profile."""
     boundaries = payload.get("operation_boundaries")
     if not isinstance(boundaries, dict):
         raise ValueError(f"{label}: operation_boundaries must be an object")
     mismatches = {
         key: boundaries.get(key)
-        for key, expected in _SAFE_OPERATION_BOUNDARIES.items()
+        for key, expected in expected_boundaries.items()
         if boundaries.get(key) is not expected
     }
     if mismatches:

apps/api/src/services/delivery_closure_workbench.py

@@ -9,6 +9,9 @@ from __future__ import annotations
 
 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_production_deploy_readback_blocker import (
     load_latest_awoooi_production_deploy_readback_blocker,
 )
@@ -46,6 +49,9 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
     production_deploy = load_latest_awoooi_production_deploy_readback_blocker()
     private_inventory = load_latest_gitea_private_inventory_p0_scorecard()
     cicd_baseline = load_latest_p0_cicd_baseline_source_readiness()
+    cicd_template_copy_apply_gate = (
+        load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate()
+    )
     gitea = load_latest_gitea_workflow_runner_health()
     runtime = load_latest_runtime_surface_inventory()
     backup = load_latest_backup_dr_readiness_matrix()
@@ -56,6 +62,7 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
         production_deploy=production_deploy,
         private_inventory=private_inventory,
         cicd_baseline=cicd_baseline,
+        cicd_template_copy_apply_gate=cicd_template_copy_apply_gate,
         gitea=gitea,
         runtime=runtime,
         backup=backup,
@@ -70,6 +77,7 @@ def build_delivery_closure_workbench(
     production_deploy: dict[str, Any],
     private_inventory: dict[str, Any],
     cicd_baseline: dict[str, Any],
+    cicd_template_copy_apply_gate: dict[str, Any],
     gitea: dict[str, Any],
     runtime: dict[str, Any],
     backup: dict[str, Any],
@@ -92,6 +100,10 @@ def build_delivery_closure_workbench(
     )
     cicd_baseline_readback = _dict(cicd_baseline.get("readback"))
     cicd_baseline_rollups = _dict(cicd_baseline.get("rollups"))
+    cicd_apply_gate_readback = _dict(cicd_template_copy_apply_gate.get("readback"))
+    cicd_apply_gate_boundaries = _dict(
+        cicd_template_copy_apply_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"))
@@ -756,6 +768,36 @@ def build_delivery_closure_workbench(
                 "workplan_id": str(
                     cicd_baseline_readback.get("workplan_id") or "P0-004"
                 ),
+                "template_copy_apply_gate_status": str(
+                    cicd_template_copy_apply_gate.get("status") or ""
+                ),
+                "template_copy_apply_allowed": (
+                    cicd_apply_gate_readback.get("apply_allowed") is True
+                ),
+                "template_copy_authorized": (
+                    cicd_apply_gate_readback.get(
+                        "workflow_template_copy_authorized"
+                    )
+                    is True
+                ),
+                "workflow_trigger_authorized": (
+                    cicd_apply_gate_readback.get("workflow_trigger_authorized")
+                    is True
+                ),
+                "auto_branch_trigger_authorized": (
+                    cicd_apply_gate_readback.get(
+                        "auto_push_or_pull_request_trigger_authorized"
+                    )
+                    is True
+                ),
+                "generic_runner_label_authorized": (
+                    cicd_apply_gate_readback.get("generic_runner_label_authorized")
+                    is True
+                ),
+                "runner_pressure_guard_required": (
+                    cicd_apply_gate_readback.get("runner_pressure_guard_required")
+                    is True
+                ),
                 "required_source_count": _int(
                     cicd_baseline_rollups.get("required_source_count")
                 ),
@@ -772,15 +814,17 @@ def build_delivery_closure_workbench(
                     cicd_baseline_rollups.get("blocked_source_ids")
                 ),
                 "workflow_modification_allowed": _dict(
-                    cicd_baseline.get("operation_boundaries")
+                    cicd_template_copy_apply_gate.get("operation_boundaries")
                 ).get("workflow_modification_allowed")
                 is True,
-                "workflow_trigger_allowed": _dict(
-                    cicd_baseline.get("operation_boundaries")
-                ).get("workflow_trigger_allowed")
+                "workflow_trigger_allowed": cicd_apply_gate_boundaries.get(
+                    "workflow_trigger_allowed"
+                )
                 is True,
                 "safe_next_step": str(
-                    cicd_baseline_readback.get("safe_next_step") or ""
+                    _first_string(cicd_template_copy_apply_gate.get("next_actions"))
+                    or cicd_baseline_readback.get("safe_next_step")
+                    or ""
                 ),
             },
             "href": "/deployments",
@@ -1120,6 +1164,33 @@ def build_delivery_closure_workbench(
             "p0_cicd_baseline_safe_next_step": str(
                 cicd_baseline_readback.get("safe_next_step") or ""
             ),
+            "p0_cicd_template_copy_apply_gate_status": str(
+                cicd_template_copy_apply_gate.get("status") or ""
+            ),
+            "p0_cicd_template_copy_apply_allowed": (
+                cicd_apply_gate_readback.get("apply_allowed") is True
+            ),
+            "p0_cicd_template_copy_authorized": (
+                cicd_apply_gate_readback.get("workflow_template_copy_authorized")
+                is True
+            ),
+            "p0_cicd_template_copy_workflow_trigger_authorized": (
+                cicd_apply_gate_readback.get("workflow_trigger_authorized") is True
+            ),
+            "p0_cicd_template_copy_auto_branch_trigger_authorized": (
+                cicd_apply_gate_readback.get(
+                    "auto_push_or_pull_request_trigger_authorized"
+                )
+                is True
+            ),
+            "p0_cicd_template_copy_generic_runner_label_authorized": (
+                cicd_apply_gate_readback.get("generic_runner_label_authorized")
+                is True
+            ),
+            "p0_cicd_template_copy_runner_pressure_guard_required": (
+                cicd_apply_gate_readback.get("runner_pressure_guard_required")
+                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"