Merge remote-tracking branch 'gitea/main' into codex/delivery-workbench-release-20260626-ffsync

apps/api/src/services/ai_agent_high_risk_owner_review_queue.py

@@ -1,10 +1,11 @@
 """
-P2-409 AI Agent high-risk owner review queue snapshot.
+P2-409 AI Agent controlled-apply / break-glass queue snapshot.
 
-Loads the latest committed high-risk owner review queue. This module only
-validates read-only approval packets, rejection guards, and reviewer checklists.
-It does not run workers, send Telegram, write Gateway queues, read secrets, call
-paid APIs, mutate hosts, run kubectl, or write production state.
+Loads the latest committed controlled-apply / critical break-glass queue. This
+module validates read-only controlled apply packets, rejection guards, and
+reviewer checklists. It does not run workers, send Telegram, write Gateway
+queues, read secrets, call paid APIs, mutate hosts, run kubectl, or write
+production state.
 """
 
 from __future__ import annotations
@@ -18,7 +19,7 @@ from src.services.snapshot_paths import default_evaluations_dir
 _DEFAULT_EVALUATIONS_DIR = default_evaluations_dir(Path(__file__))
 _SNAPSHOT_PATTERN = "ai_agent_high_risk_owner_review_queue_*.json"
 _SCHEMA_VERSION = "ai_agent_high_risk_owner_review_queue_v1"
-_RUNTIME_AUTHORITY = "high_risk_owner_review_queue_no_live_execution_committed_snapshot"
+_RUNTIME_AUTHORITY = "controlled_apply_break_glass_queue_readback_no_live_execution"
 _EXPECTED_CURRENT_TASK = "P2-409"
 _EXPECTED_NEXT_TASK = "P2-410"
 _EXPECTED_CANONICAL_ROOM = "AwoooI SRE 戰情室"
@@ -38,13 +39,15 @@ _TRUE_TRUTH_FLAGS = {
     "p2_110e_work_items_owner_review_loaded",
     "telegram_egress_inventory_loaded",
     "telegram_owner_request_draft_loaded",
-    "all_high_risk_actions_paused",
     "approval_packets_ready",
     "rejection_guards_ready",
     "reviewer_checklists_ready",
-    "high_risk_owner_review_required",
+    "high_risk_controlled_apply_enabled",
+    "critical_break_glass_required",
 }
 _FALSE_TRUTH_FLAGS = {
+    "all_high_risk_actions_paused",
+    "high_risk_owner_review_required",
     "auto_worker_enabled",
     "live_execution_enabled",
     "gateway_queue_write_enabled",
@@ -77,12 +80,14 @@ _ZERO_TRUTH_COUNTS = {
     "redacted_payload_ingested_count_24h",
 }
 _TRUE_BOUNDARY_FLAGS = {
-    "read_only_owner_review_queue_allowed",
+    "controlled_apply_queue_readback_allowed",
+    "critical_break_glass_queue_readback_allowed",
     "approval_packet_preview_allowed",
     "rejection_guard_preview_allowed",
     "reviewer_checklist_allowed",
 }
 _FALSE_BOUNDARY_FLAGS = {
+    "read_only_owner_review_queue_allowed",
     "auto_worker_enabled",
     "live_execution_enabled",
     "gateway_queue_write_enabled",
@@ -133,11 +138,11 @@ _FORBIDDEN_PUBLIC_TERMS = {
 def load_latest_ai_agent_high_risk_owner_review_queue(
     evaluations_dir: Path | None = None,
 ) -> dict[str, Any]:
-    """Load the newest committed P2-409 high-risk owner review queue snapshot."""
+    """Load the newest committed P2-409 controlled apply queue snapshot."""
     directory = evaluations_dir or _DEFAULT_EVALUATIONS_DIR
     candidates = sorted(directory.glob(_SNAPSHOT_PATTERN))
     if not candidates:
-        raise FileNotFoundError(f"no AI Agent high-risk owner review queue snapshots found in {directory}")
+        raise FileNotFoundError(f"no AI Agent controlled apply queue snapshots found in {directory}")
 
     latest = candidates[-1]
     with latest.open(encoding="utf-8") as handle:
@@ -222,7 +227,8 @@ def _require_queue_items(payload: dict[str, Any], label: str) -> None:
         if item.get("risk_tier") not in {"high", "critical"}:
             raise ValueError(f"{label}: queue item {item_id} must be high or critical")
         if item.get("queue_status") not in {
-            "paused_owner_review_required",
+            "controlled_apply_packet_ready",
+            "critical_break_glass_required",
             "blocked_missing_owner_response",
             "approval_packet_preview_ready",
         }:
@@ -238,7 +244,13 @@ def _require_queue_items(payload: dict[str, Any], label: str) -> None:
         ):
             if not item.get(field):
                 raise ValueError(f"{label}: queue item {item_id} missing {field}")
-        for flag in ("owner_response_required", "rollback_owner_required", "postcheck_required"):
+        expected_owner_response = item.get("risk_tier") == "critical"
+        if item.get("owner_response_required") is not expected_owner_response:
+            raise ValueError(
+                f"{label}: queue item {item_id}.owner_response_required must be "
+                f"{expected_owner_response}"
+            )
+        for flag in ("rollback_owner_required", "postcheck_required"):
             if item.get(flag) is not True:
                 raise ValueError(f"{label}: queue item {item_id}.{flag} must remain true")
         for flag in (
@@ -265,7 +277,11 @@ def _require_approval_packets(payload: dict[str, Any], label: str) -> None:
         packet_id = packet.get("approval_packet_id") or "<missing>"
         if packet.get("queue_item_id") not in queue_ids:
             raise ValueError(f"{label}: approval packet {packet_id} references unknown queue item")
-        if packet.get("packet_status") not in {"draft_ready_owner_response_required", "blocked_missing_owner_response"}:
+        if packet.get("packet_status") not in {
+            "controlled_apply_packet_ready",
+            "break_glass_packet_ready",
+            "blocked_missing_owner_response",
+        }:
             raise ValueError(f"{label}: approval packet {packet_id} status is invalid")
         for field in ("required_owner_fields", "required_evidence_refs", "reviewer_checklist_id", "rejection_guard_ids"):
             if not packet.get(field):
@@ -334,10 +350,10 @@ def _require_reviewer_checklists(payload: dict[str, Any], label: str) -> None:
 def _require_routing_policy(payload: dict[str, Any], label: str) -> None:
     policy = payload.get("routing_policy") or {}
     expected = {
-        "high_risk_default_route": "pause_to_owner_review_queue",
-        "critical_risk_default_route": "pause_to_owner_review_queue",
-        "low_medium_runtime_route": "pause_until_owner_approved_runtime_gate",
-        "owner_response_required": True,
+        "high_risk_default_route": "controlled_apply_queue",
+        "critical_risk_default_route": "critical_break_glass_queue",
+        "low_medium_runtime_route": "controlled_apply_queue",
+        "owner_response_required": False,
         "verbal_approval_accepted": False,
         "redacted_payload_only": True,
     }
@@ -414,6 +430,10 @@ def _require_rollups(payload: dict[str, Any], label: str) -> None:
         "rollback_owner_required_count": sum(1 for item in items if item.get("rollback_owner_required") is True),
         "postcheck_required_count": sum(1 for item in items if item.get("postcheck_required") is True),
         "blocked_runtime_action_count": len(blocked_actions),
+        "controlled_apply_queue_count": sum(1 for item in items if item.get("risk_tier") == "high"),
+        "critical_break_glass_queue_count": sum(1 for item in items if item.get("risk_tier") == "critical"),
+        "owner_response_required_count": sum(1 for item in items if item.get("owner_response_required") is True),
+        "high_risk_owner_review_required_count": 0,
     }
     mismatches = {
         key: {"expected": value, "actual": rollups.get(key)}