diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py
index d7e12cbf3..e7605a80c 100644
--- a/apps/api/src/api/v1/webhooks.py
+++ b/apps/api/src/api/v1/webhooks.py
@@ -44,7 +44,6 @@ from src.models.approval import (
RiskLevel,
)
from src.models.webhook import AlertPayload, AlertResponse
-from src.services.action_parser import is_safe_kubectl_action
from src.services.alert_analyzer_service import AlertAnalyzer
from src.services.alert_approval_guard import guard_alert_approval_action
from src.services.alert_grouping_service import get_alert_grouping_service
@@ -217,6 +216,18 @@ def _auto_repair_action_label(result, fallback_target: str) -> str:
return f"auto_repair_playbook:{playbook_id} {step_text}".strip()
+def _controlled_ai_policy_allows(risk_level: RiskLevel | str) -> bool:
+ """Return whether the current owner policy waives human approval.
+
+ The decision only opens the single-writer check/apply/verifier route. It
+ does not bypass catalog allowlists, check-mode, rollback, or post-verifier
+ requirements. Critical actions remain break-glass only.
+ """
+
+ value = risk_level.value if isinstance(risk_level, RiskLevel) else str(risk_level)
+ return value.strip().lower() in {"low", "medium", "high"}
+
+
async def _try_auto_repair_background(
incident_id: str,
approval_id: str,
@@ -1247,6 +1258,13 @@ async def receive_alert(
)
if existing_approval:
+ policy_promoted = False
+ promote = getattr(service, "apply_current_owner_policy", None)
+ if callable(promote):
+ promoted_approval, policy_promoted = await promote(existing_approval.id)
+ if promoted_approval is not None:
+ existing_approval = promoted_approval
+
# ==========================================================================
# 戰略 B Step 3: [收斂] 同指紋告警 - 跳過 LLM，只更新計數！
# ==========================================================================
@@ -1263,6 +1281,21 @@ async def receive_alert(
updated_approval = await service.increment_hit_count(existing_approval.id)
if updated_approval:
+ if (
+ policy_promoted
+ and getattr(updated_approval, "incident_id", None)
+ and _controlled_ai_policy_allows(updated_approval.risk_level)
+ and not is_heartbeat_alertname(alert.alert_type)
+ ):
+ background_tasks.add_task(
+ _try_auto_repair_background,
+ incident_id=str(updated_approval.incident_id),
+ approval_id=str(updated_approval.id),
+ alert_type=alert.alert_type,
+ target_resource=alert.target_resource,
+ namespace=alert.namespace,
+ risk_level=updated_approval.risk_level.value,
+ )
# =================================================================
# 2026-03-27 ogt: 收斂告警不重複發送 Telegram，只更新 hit_count
# 避免 Telegram 洗版，用戶可在 UI 查看聚合次數
@@ -1363,6 +1396,7 @@ async def receive_alert(
risk_mapping = {
"low": RiskLevel.LOW,
"medium": RiskLevel.MEDIUM,
+ "high": RiskLevel.HIGH,
"critical": RiskLevel.CRITICAL,
}
risk_level = risk_mapping.get(analysis_result.risk_level.value, RiskLevel.MEDIUM)
@@ -1500,27 +1534,20 @@ async def receive_alert(
except Exception as _shadow_err:
logger.warning("shadow_auto_approve_failed", error=str(_shadow_err))
- # 2026-04-27 ogt + Claude Sonnet 4.6: CS1 LLM 高信心度自動執行
- # 設計：confidence ≥ 0.85 + 非 CRITICAL + 非破壞性 + 有 kubectl 指令 → 直接執行
- # 安全防線：CRITICAL / destructive patterns / NO_ACTION/INVESTIGATE/OBSERVE / 空 kubectl → 降級 PENDING
- if analysis_result:
- _cs1_kubectl = _cmd_cs1
- _cs1_can_auto = (
- bool(_cs1_kubectl)
- and analysis_result.confidence >= 0.85
- and _cs1_risk_level != RiskLevel.CRITICAL
- and _sa_val not in _non_destructive_actions
- and is_safe_kubectl_action(_cs1_kubectl)
+ # Current policy routes every non-critical alert to the single-writer
+ # executor. Confidence and command shape remain evidence for check-mode;
+ # they are not human-approval gates.
+ if _controlled_ai_policy_allows(_cs1_risk_level) and not is_heartbeat_alertname(
+ str((alert.labels or {}).get("alertname") or alert.alert_type or "")
+ ):
+ await _try_auto_repair_background(
+ incident_id=_cs1_incident_id,
+ approval_id=str(approval.id),
+ alert_type=alert.alert_type,
+ target_resource=alert.target_resource,
+ namespace=alert.namespace,
+ risk_level=_cs1_risk_level.value,
)
- if _cs1_can_auto:
- await _try_auto_repair_background(
- incident_id=_cs1_incident_id,
- approval_id=str(approval.id),
- alert_type=alert.alert_type,
- target_resource=alert.target_resource,
- namespace=alert.namespace,
- risk_level=_cs1_risk_level.value,
- )
logger.info(
"approval_auto_created_with_fingerprint",
@@ -1749,6 +1776,7 @@ async def _process_new_alert_background(
risk_mapping = {
"low": RiskLevel.LOW,
"medium": RiskLevel.MEDIUM,
+ "high": RiskLevel.HIGH,
"critical": RiskLevel.CRITICAL,
}
rule_risk = risk_mapping.get(
@@ -1860,12 +1888,7 @@ async def _process_new_alert_background(
# 2026-04-27 ogt + Claude Sonnet 4.6: CS2 規則引擎自動執行
# 設計：is_rule_based=True 確定性高，滿足條件直接執行，不等人工審核
# 安全防線：CRITICAL / destructive patterns / NO_ACTION / 空 kubectl → 全部降級 PENDING
- _can_auto = (
- bool(rule_kubectl)
- and rule_risk != RiskLevel.CRITICAL
- and is_safe_kubectl_action(rule_kubectl)
- and "NO_ACTION" not in rule_action
- )
+ _can_auto = _controlled_ai_policy_allows(rule_risk)
if _can_auto:
logger.info(
"rule_engine_auto_execution_deferred_to_single_writer",
@@ -1919,7 +1942,7 @@ async def _process_new_alert_background(
)
_is_heartbeat = is_heartbeat_alertname(alertname)
- if can_auto_repair and not _is_heartbeat:
+ if _controlled_ai_policy_allows(rule_risk) and not _is_heartbeat:
await _try_auto_repair_background(
incident_id=incident_id,
approval_id=str(approval.id),
@@ -1928,7 +1951,7 @@ async def _process_new_alert_background(
namespace=namespace,
risk_level=rule_risk.value,
)
- elif not can_auto_repair and not _is_heartbeat:
+ elif not _is_heartbeat:
from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository
_op_log_rule = get_alert_operation_log_repository()
await _op_log_rule.append(
@@ -1936,9 +1959,13 @@ async def _process_new_alert_background(
incident_id=incident_id,
approval_id=str(approval.id),
actor="prometheus-rule",
- action_detail=f"Prometheus rule 設定 auto_repair=false，轉入緊急介入: {alertname}",
+ action_detail=f"Critical alert requires break-glass: {alertname}",
success=False,
- context={"alertname": alertname, "auto_repair_flag": False},
+ context={
+ "alertname": alertname,
+ "legacy_auto_repair_flag": bool(can_auto_repair),
+ "critical_break_glass": True,
+ },
)
await _escalate_auto_repair_unavailable(
incident_id=incident_id,
@@ -1946,8 +1973,8 @@ async def _process_new_alert_background(
alert_type=alert_type,
target_resource=target_resource,
namespace=namespace,
- failure_reason="Prometheus rule auto_repair=false，未進入自動修復評估",
- attempted_actions="rule_engine -> rule_first -> emergency_intervention",
+ failure_reason="critical risk requires break-glass authorization",
+ attempted_actions="rule_engine -> critical_break_glass_gate",
)
await _push_to_telegram_background(
@@ -1983,6 +2010,7 @@ async def _process_new_alert_background(
risk_mapping = {
"low": RiskLevel.LOW,
"medium": RiskLevel.MEDIUM,
+ "high": RiskLevel.HIGH,
"critical": RiskLevel.CRITICAL,
}
risk_level = risk_mapping.get(analysis_result.risk_level.value, RiskLevel.MEDIUM)
@@ -2078,15 +2106,8 @@ async def _process_new_alert_background(
except Exception as _shadow_err_cs3:
logger.warning("shadow_auto_approve_failed", error=str(_shadow_err_cs3))
- # 2026-04-27 Claude Sonnet 4.6: CS3 LLM 高信心自動執行（修法3擴展）
_cs3_kubectl = _cmd_cs3
- _cs3_can_auto = (
- bool(_cs3_kubectl)
- and analysis_result.confidence >= 0.85
- and risk_level != RiskLevel.CRITICAL
- and "NO_ACTION" not in (analysis_result.action_title or "")
- and is_safe_kubectl_action(_cs3_kubectl)
- )
+ _cs3_can_auto = _controlled_ai_policy_allows(risk_level)
if _cs3_can_auto:
logger.info(
"cs3_llm_auto_execution_deferred_to_single_writer",
@@ -2153,7 +2174,7 @@ async def _process_new_alert_background(
alertname=alertname,
)
- if can_auto_repair and not _is_heartbeat:
+ if _controlled_ai_policy_allows(risk_level) and not _is_heartbeat:
await _try_auto_repair_background(
incident_id=incident_id,
approval_id=str(approval.id),
@@ -2162,7 +2183,7 @@ async def _process_new_alert_background(
namespace=namespace,
risk_level=risk_level.value,
)
- else:
+ elif not _is_heartbeat:
from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository
_op_log_rule = get_alert_operation_log_repository()
await _op_log_rule.append(
@@ -2170,20 +2191,23 @@ async def _process_new_alert_background(
incident_id=incident_id,
approval_id=str(approval.id),
actor="prometheus-rule",
- action_detail=f"Prometheus rule 設定 auto_repair=false，強制人工審核: {alertname}",
+ action_detail=f"Critical alert requires break-glass: {alertname}",
success=False,
- context={"alertname": alertname, "auto_repair_flag": False},
+ context={
+ "alertname": alertname,
+ "legacy_auto_repair_flag": bool(can_auto_repair),
+ "critical_break_glass": True,
+ },
+ )
+ await _escalate_auto_repair_unavailable(
+ incident_id=incident_id,
+ approval_id=str(approval.id),
+ alert_type=alert_type,
+ target_resource=target_resource,
+ namespace=namespace,
+ failure_reason="critical risk requires break-glass authorization",
+ attempted_actions="llm_analysis -> critical_break_glass_gate",
)
- if not _is_heartbeat:
- await _escalate_auto_repair_unavailable(
- incident_id=incident_id,
- approval_id=str(approval.id),
- alert_type=alert_type,
- target_resource=target_resource,
- namespace=namespace,
- failure_reason="Prometheus rule auto_repair=false，未進入自動修復評估",
- attempted_actions="llm_analysis -> guardrail:auto_repair_false -> emergency_intervention",
- )
await _push_to_telegram_background(
approval_id=str(approval.id),
@@ -2413,6 +2437,15 @@ async def _process_new_alert_background(
from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository
_op_log_fallback = get_alert_operation_log_repository()
if repair_candidate_result.candidate_found:
+ if _controlled_ai_policy_allows(fallback_create.risk_level):
+ await _try_auto_repair_background(
+ incident_id=fallback_incident_id,
+ approval_id=str(approval.id),
+ alert_type=alert_type,
+ target_resource=target_resource,
+ namespace=namespace,
+ risk_level=fallback_create.risk_level.value,
+ )
await _op_log_fallback.append(
"PLAYBOOK_DRAFT_CREATED",
incident_id=fallback_incident_id,
@@ -2620,7 +2653,8 @@ async def alertmanager_webhook(
_alert_labels = alert.labels or {}
_alert_annotations = alert.annotations or {}
_alertname_for_log = _alert_labels.get("alertname", "UnknownAlert")
- # Q9: auto_repair flag — Rule=false 強制 HITL（不觸發自動修復背景任務）
+ # Legacy rule flag is retained as evidence only. Current owner policy routes
+ # low/medium/high alerts through the controlled single-writer pipeline.
_can_auto_repair_by_rule = _alert_labels.get("auto_repair", "true").lower() == "true"
try:
from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository
@@ -2878,6 +2912,13 @@ async def alertmanager_webhook(
)
if existing_approval:
+ policy_promoted = False
+ promote = getattr(service, "apply_current_owner_policy", None)
+ if callable(promote):
+ promoted_approval, policy_promoted = await promote(existing_approval.id)
+ if promoted_approval is not None:
+ existing_approval = promoted_approval
+
# 收斂告警 - 跳過 LLM
logger.info(
"alertmanager_converged",
@@ -2888,6 +2929,21 @@ async def alertmanager_webhook(
updated_approval = await service.increment_hit_count(existing_approval.id)
if updated_approval:
+ if (
+ policy_promoted
+ and getattr(updated_approval, "incident_id", None)
+ and _controlled_ai_policy_allows(updated_approval.risk_level)
+ and not is_heartbeat_alertname(alertname)
+ ):
+ background_tasks.add_task(
+ _try_auto_repair_background,
+ incident_id=str(updated_approval.incident_id),
+ approval_id=str(updated_approval.id),
+ alert_type=alert_type,
+ target_resource=target_resource,
+ namespace=namespace,
+ risk_level=updated_approval.risk_level.value,
+ )
# 2026-03-27 ogt: 收斂告警不重複發送 Telegram，只更新 hit_count
# 用戶可在 UI 查看聚合次數，避免 Telegram 洗版
logger.info(
diff --git a/apps/api/src/core/trust_engine.py b/apps/api/src/core/trust_engine.py
index 4f6b038dc..abc4015a9 100644
--- a/apps/api/src/core/trust_engine.py
+++ b/apps/api/src/core/trust_engine.py
@@ -4,9 +4,8 @@ Trust Engine - 風險判定與 Multi-Sig 簽核邏輯
CISO-101: 信任引擎核心實作
風險等級與簽核需求:
-- LOW: 0 人，自動放行 (如 scale up)
-- MEDIUM: 需 1 人簽核 (如 delete pod)
-- CRITICAL: 需 2 人 Multi-Sig 雙重簽核 (如 DROP TABLE)
+- LOW/MEDIUM/HIGH: 0 人，進入受控自動執行
+- CRITICAL: 1 人 break-glass (如 DROP TABLE / host reboot / secret read)
Features:
- 自動風險分類
@@ -42,6 +41,37 @@ CRITICAL_KEYWORDS = [
"format",
"wipe",
"purge all",
+ "destructive migration",
+ "restore database",
+ "restore backup",
+ "remote delete",
+ "retention delete",
+ "retention prune",
+ "host reboot",
+ "reboot host",
+ "node drain",
+ "firewall cutover",
+ "dns cutover",
+ "credentialed exploit",
+ "active scan",
+ "paid provider",
+ "cost limit",
+ "switch production provider",
+ "replace production model",
+ "replace agent runtime",
+ "force push",
+ "delete repo",
+ "delete ref",
+ "change repo visibility",
+ "raw runtime secret volume",
+ "kubectl get secret",
+ "kubectl describe secret",
+ "read secret",
+ "dump secret",
+ "export token",
+ "read token",
+ "private key",
+ "authorization header",
]
MEDIUM_KEYWORDS = [
@@ -69,9 +99,10 @@ LOW_KEYWORDS = [
# =============================================================================
SIGNATURE_REQUIREMENTS: dict[RiskLevel, int] = {
- RiskLevel.LOW: 0, # 自動放行
- RiskLevel.MEDIUM: 1, # 單人簽核
- RiskLevel.CRITICAL: 1, # 2026-04-02 ogt: 統帥決策 — 只需 1 層審核，Multi-Sig 停用
+ RiskLevel.LOW: 0,
+ RiskLevel.MEDIUM: 0,
+ RiskLevel.HIGH: 0,
+ RiskLevel.CRITICAL: 1,
}
@@ -162,10 +193,6 @@ def classify_risk(
Returns:
最終風險等級
"""
- # 如果明確指定，直接使用
- if explicit_level is not None:
- return explicit_level
-
# 從動作分類
action_risk = classify_risk_by_action(action)
@@ -175,11 +202,12 @@ def classify_risk(
blast_risk = classify_risk_by_blast_radius(blast_radius)
# 取較高風險等級
- risk_order = [RiskLevel.LOW, RiskLevel.MEDIUM, RiskLevel.CRITICAL]
+ risk_order = [RiskLevel.LOW, RiskLevel.MEDIUM, RiskLevel.HIGH, RiskLevel.CRITICAL]
action_idx = risk_order.index(action_risk)
blast_idx = risk_order.index(blast_risk)
+ explicit_idx = risk_order.index(explicit_level) if explicit_level is not None else 0
- return risk_order[max(action_idx, blast_idx)]
+ return risk_order[max(action_idx, blast_idx, explicit_idx)]
# =============================================================================
@@ -220,7 +248,7 @@ class TrustEngine:
建立新的授權請求
自動根據風險等級設定所需簽核數
- LOW 風險自動批准
+ LOW/MEDIUM/HIGH 風險由 current owner policy 自動批准
"""
# 分類風險
risk_level = classify_risk(
@@ -245,8 +273,8 @@ class TrustEngine:
required_signatures=required_sigs,
)
- # LOW 風險自動批准
- if risk_level == RiskLevel.LOW:
+ # Non-critical requests enter controlled automation without a human gate.
+ if required_sigs == 0:
approval.status = ApprovalStatus.APPROVED
approval.resolved_at = datetime.now(UTC)
if self._on_approved:
diff --git a/apps/api/src/models/approval.py b/apps/api/src/models/approval.py
index 6a6ce1a9e..2761b35b8 100644
--- a/apps/api/src/models/approval.py
+++ b/apps/api/src/models/approval.py
@@ -5,8 +5,8 @@ CISO-101: 授權請求與簽核資料模型
Features:
- 狀態機 (PENDING → APPROVED/REJECTED/EXPIRED)
-- 風險等級判定 (LOW/MEDIUM/CRITICAL)
-- Multi-Sig 簽核追蹤
+- 風險等級判定 (LOW/MEDIUM/HIGH/CRITICAL)
+- Controlled-apply / break-glass 簽核追蹤
- Pydantic 強型別驗證
"""
@@ -41,10 +41,9 @@ class RiskLevel(str, Enum):
"""
風險等級 - 決定所需簽核人數
- - LOW: 0 人，自動放行
- - MEDIUM: 需 1 人簽核
- - HIGH: 需 1 人簽核 (信任引擎可降級至 MEDIUM)
- - CRITICAL: 需 2 人 Multi-Sig 雙重簽核 (永不降級)
+ - LOW/MEDIUM/HIGH: current owner policy 自動授權，後續仍須
+ check-mode、bounded execution、獨立 verifier 與 rollback receipt
+ - CRITICAL: break-glass 人工授權 (永不自動降級)
變更紀錄:
- 2026-03-25: 新增 HIGH (Phase 16 R2 合併自 trust_engine.py)
diff --git a/apps/api/src/services/approval_db.py b/apps/api/src/services/approval_db.py
index b2e664bf4..5328c0ab9 100644
--- a/apps/api/src/services/approval_db.py
+++ b/apps/api/src/services/approval_db.py
@@ -159,7 +159,7 @@ def approval_request_to_record_data(
record_data = {
"action": request.action,
"description": request.description,
- "status": ApprovalStatus.APPROVED if risk_level == RiskLevel.LOW else ApprovalStatus.PENDING,
+ "status": ApprovalStatus.APPROVED if required_sigs == 0 else ApprovalStatus.PENDING,
"risk_level": risk_level,
"required_signatures": required_sigs,
"current_signatures": 0,
@@ -169,7 +169,7 @@ def approval_request_to_record_data(
"requested_by": request.requested_by,
"expires_at": request.expires_at,
"extra_metadata": metadata or None,
- "resolved_at": now if risk_level == RiskLevel.LOW else None,
+ "resolved_at": now if required_sigs == 0 else None,
# 戰略 B: 告警風暴收斂
"fingerprint": fingerprint,
"hit_count": 1,
@@ -279,7 +279,7 @@ def _approval_decision_mode(record: ApprovalRecord) -> str:
risk_level = _record_value(record.risk_level)
if _approval_needs_human(record) or current > 0:
return "manual"
- if risk_level == "low" and required == 0:
+ if risk_level in {"low", "medium", "high"} and required == 0:
return "auto"
return "manual"
@@ -442,6 +442,66 @@ class ApprovalDBService:
return approval_record_to_request(record)
+ async def apply_current_owner_policy(
+ self,
+ approval_id: UUID,
+ ) -> tuple[ApprovalRequest | None, bool]:
+ """Promote a legacy pending non-critical gate to controlled automation.
+
+ Existing alert fingerprints can otherwise keep converging onto an old
+ PENDING row for 24 hours. This bounded reconciliation updates only one
+ row, re-runs the current critical classifier, and leaves critical rows
+ untouched for break-glass handling.
+ """
+
+ async with get_db_context() as db:
+ result = await db.execute(
+ select(ApprovalRecord)
+ .where(ApprovalRecord.id == str(approval_id))
+ .with_for_update()
+ )
+ record = result.scalar_one_or_none()
+ if record is None:
+ return None, False
+
+ current = approval_record_to_request(record)
+ status = _record_value(record.status)
+ effective_risk = classify_risk(
+ action=str(record.action or ""),
+ blast_radius=current.blast_radius,
+ explicit_level=current.risk_level,
+ )
+ if status != "pending" or effective_risk == RiskLevel.CRITICAL:
+ return current, False
+
+ metadata = dict(record.extra_metadata or {})
+ metadata.update(
+ {
+ "owner_policy": "global_product_governance_v2",
+ "owner_review_gate": "auto_waived_for_low_medium_high",
+ "controlled_apply_required": True,
+ "legacy_pending_promoted": True,
+ }
+ )
+ record.status = ApprovalStatus.APPROVED
+ record.risk_level = effective_risk
+ record.required_signatures = 0
+ record.current_signatures = 0
+ record.signatures = []
+ record.resolved_at = datetime.now(UTC)
+ record.extra_metadata = metadata
+ add_approval_created_timeline_event(db, record)
+ await db.flush()
+ await db.refresh(record)
+
+ logger.info(
+ "approval_promoted_by_current_owner_policy",
+ id=record.id,
+ risk_level=effective_risk.value,
+ incident_id=record.incident_id,
+ )
+ return approval_record_to_request(record), True
+
async def find_by_fingerprint(
self,
fingerprint: str,
diff --git a/apps/api/src/services/operator_outcome.py b/apps/api/src/services/operator_outcome.py
index 300872b31..ac92fcda4 100644
--- a/apps/api/src/services/operator_outcome.py
+++ b/apps/api/src/services/operator_outcome.py
@@ -143,7 +143,7 @@ def normalize_operator_outcome(outcome: dict[str, Any] | None) -> dict[str, Any]
if isinstance(normalized.get("notification"), dict)
else {}
)
- notification["mode"] = "action_required"
+ notification["mode"] = "human_action_required" if break_glass else "automation_progress"
notification["channels"] = list(_ACTION_REQUIRED_CHANNELS)
notification["reason"] = str(normalized["human_action_reason"])
normalized["notification"] = notification
@@ -200,11 +200,10 @@ def _build_notification(
"channels": list(channels),
"reason": reason,
"source_id": source_id,
- "telegram": (
- "reply_to_original_or_standalone_action_required"
- if mode == "action_required"
- else "reply_to_original_or_standalone_result"
- ),
+ "telegram": {
+ "human_action_required": "reply_to_original_or_standalone_action_required",
+ "automation_progress": "reply_to_original_or_standalone_automation_progress",
+ }.get(mode, "reply_to_original_or_standalone_result"),
"awooop": "status_chain_panel",
}
@@ -607,8 +606,13 @@ def build_operator_outcome(
"monitor_for_regression",
"monitor_or_reopen_if_alert_recurs",
}
- mode = "action_required" if needs_human or ai_action_required else "result_only"
- channels = _ACTION_REQUIRED_CHANNELS if mode == "action_required" else _RESULT_ONLY_CHANNELS
+ if needs_human:
+ mode = "human_action_required"
+ elif ai_action_required:
+ mode = "automation_progress"
+ else:
+ mode = "result_only"
+ channels = _ACTION_REQUIRED_CHANNELS if mode != "result_only" else _RESULT_ONLY_CHANNELS
return {
"schema_version": "operator_outcome_v1",
"state": state,
diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py
index 774de7eb1..5ddad0403 100644
--- a/apps/api/src/services/telegram_gateway.py
+++ b/apps/api/src/services/telegram_gateway.py
@@ -3374,6 +3374,7 @@ class TelegramMessage:
repair_candidate_promotion_summary: str = "" # 候選升級合約摘要
repair_candidate_work_item_href: str = "" # AwoooP 修復候選草案連結
repair_candidate_work_item_id: str = "" # AwoooP 修復候選草案 ID
+ controlled_auto_authorized: bool = False # low/medium/high owner policy
# ==========================================================================
# Phase 22: Nemotron 協作欄位 (ADR-044)
@@ -3529,7 +3530,11 @@ class TelegramMessage:
nemotron_state = "tool_ready" if self.nemotron_enabled else "standby"
hermes_state = self.playbook_name or "rule_catalog"
elephant_state = "timeline_km_pending"
- flow = "webhook>investigator>router>llm/rule>safe>approval"
+ flow = (
+ "webhook>investigator>router>policy>check>apply>verify>writeback"
+ if self.controlled_auto_authorized
+ else "webhook>investigator>router>safe>break-glass"
+ )
# 2026-05-04 ogt: Token 用量顯示（有資料才顯示）
token_line = ""
@@ -3608,8 +3613,10 @@ class TelegramMessage:
execute_state = f"diagnostic_recorded:{operation_records}"
elif is_noop:
execute_state = "no_action_or_observe"
+ elif self.controlled_auto_authorized:
+ execute_state = "controlled_queue_evaluation"
elif "approval" in verdict or self._automation_mode() == "ai_proposal_ready":
- execute_state = "awaiting_approval"
+ execute_state = "awaiting_break_glass"
else:
execute_state = "not_started"
@@ -3628,8 +3635,10 @@ class TelegramMessage:
conclusion = "已記錄診斷/觀察，尚未證明修復"
elif is_noop:
conclusion = "未自動修復，已排入 AI 證據補強 / PlayBook 候選"
+ elif self.controlled_auto_authorized:
+ conclusion = "已由 owner policy 自動授權，等待 check/apply/verifier receipt"
elif "approval" in verdict:
- conclusion = "等待 controlled policy 判定後執行"
+ conclusion = "critical break-glass 尚未授權"
elif "manual" in verdict:
conclusion = "舊人工狀態已轉入 AI 受控處理"
else:
@@ -3811,8 +3820,13 @@ class TelegramMessage:
manual_handoff_block = self._format_manual_handoff_package_block()
# ADR-075 TYPE-3 格式組裝
+ title = (
+ "AI CONTROLLED AUTOMATION"
+ if self.controlled_auto_authorized
+ else "BREAK-GLASS REQUIRED"
+ )
message = (
- f"{self.status_emoji} ACTION REQUIRED | {html.escape(risk_zh)}\n"
+ f"{self.status_emoji} {title} | {html.escape(risk_zh)}\n"
f"──────────────────────\n"
f"📋 {html.escape(incident_id)}\n"
f"🎯 資源：{safe_resource}\n"
@@ -3871,6 +3885,9 @@ class TelegramMessage:
Returns:
str: 格式化的 Telegram 訊息 (max 1000 字元)
"""
+ if self.controlled_auto_authorized:
+ return self.format()
+
# 責任映射
resp_map = {
"FE": "👨💻 FE (前端)",
@@ -5358,6 +5375,7 @@ class TelegramGateway:
action_plan: object = None,
repair_candidate_work_item_href: str = "",
repair_candidate_work_item_id: str = "",
+ controlled_auto_authorized: bool = False,
) -> dict:
"""
建立 Inline Keyboard
@@ -5395,6 +5413,35 @@ class TelegramGateway:
else is_executable_repair_approval_action(suggested_action)
)
+ if controlled_auto_authorized:
+ rows: list[list[dict]] = []
+ if incident_id:
+ first_row: list[dict] = []
+ work_item_deep_link = _repair_candidate_work_item_url(
+ incident_id=incident_id,
+ repair_candidate_work_item_href=repair_candidate_work_item_href,
+ repair_candidate_work_item_id=repair_candidate_work_item_id,
+ )
+ if work_item_deep_link:
+ first_row.append({"text": "🧾 Work Item", "url": work_item_deep_link})
+ first_row.extend(
+ [
+ {"text": "🧰 處置包", "callback_data": f"detail:{incident_id}"},
+ {"text": "📊 歷史", "callback_data": f"history:{incident_id}"},
+ ]
+ )
+ rows.append(first_row)
+ awooop_row = _awooop_truth_chain_button_row(incident_id)
+ if awooop_row:
+ rows.append(awooop_row)
+ logger.info(
+ "telegram_keyboard_built",
+ source="controlled_auto_authorized",
+ approval_id=approval_id,
+ incident_id=incident_id,
+ )
+ return {"inline_keyboard": rows}
+
# 可執行修復卡第一排置頂批准/拒絕；純觀察卡不得提供誤導性的執行批准。
first_row: list[dict] = [
{"text": "✅ 批准", "callback_data": approve_nonce},
@@ -5782,6 +5829,12 @@ class TelegramGateway:
Returns:
dict: Telegram API 回應
"""
+ controlled_auto_authorized = risk_level.strip().lower() in {
+ "low",
+ "medium",
+ "high",
+ }
+
# 取得狀態 Emoji
emoji = RISK_EMOJI_MAP.get(risk_level.lower(), "⚠️")
@@ -5869,6 +5922,7 @@ class TelegramGateway:
repair_candidate_promotion_summary=repair_candidate_promotion_summary,
repair_candidate_work_item_href=repair_candidate_work_item_href,
repair_candidate_work_item_id=repair_candidate_work_item_id,
+ controlled_auto_authorized=controlled_auto_authorized,
)
# 格式化訊息 — Phase 22: 如果 Nemotron 啟用，使用雙軌格式
@@ -5887,6 +5941,7 @@ class TelegramGateway:
notification_type=notification_type,
repair_candidate_work_item_href=repair_candidate_work_item_href,
repair_candidate_work_item_id=repair_candidate_work_item_id,
+ controlled_auto_authorized=controlled_auto_authorized,
)
# 發送訊息：2026-04-30 統帥指示，告警卡片完整切到 SRE 戰情室群組。
@@ -6030,6 +6085,11 @@ class TelegramGateway:
由 asyncio.create_task 非同步呼叫，失敗不影響主告警流程。
"""
try:
+ controlled_auto_authorized = risk_level.strip().lower() in {
+ "low",
+ "medium",
+ "high",
+ }
emoji = RISK_EMOJI_MAP.get(risk_level.lower(), "⚠️")
signoz_metrics = None
@@ -6072,6 +6132,7 @@ class TelegramGateway:
nemotron_validation=nemotron_validation,
nemotron_latency_ms=nemotron_latency_ms,
remediation_summary=remediation_summary,
+ controlled_auto_authorized=controlled_auto_authorized,
)
text = message.format_with_nemotron() if nemotron_enabled else message.format()
@@ -6083,6 +6144,7 @@ class TelegramGateway:
suggested_action=suggested_action,
alert_category=alert_category,
notification_type=notification_type,
+ controlled_auto_authorized=controlled_auto_authorized,
)
resp = await self.send_to_group(text=text, reply_markup=_group_keyboard)
diff --git a/apps/api/tests/test_awooop_operator_timeline_labels.py b/apps/api/tests/test_awooop_operator_timeline_labels.py
index 87dab48bf..811b8dfba 100644
--- a/apps/api/tests/test_awooop_operator_timeline_labels.py
+++ b/apps/api/tests/test_awooop_operator_timeline_labels.py
@@ -2781,7 +2781,7 @@ def test_awooop_status_chain_does_not_treat_audit_ops_as_repair() -> None:
assert chain["next_step"] == "auto_generate_repair_candidate_from_diagnostic_evidence"
assert chain["needs_human"] is False
assert chain["operator_outcome"]["state"] == "diagnostic_only_ai_repair_required"
- assert chain["operator_outcome"]["notification"]["mode"] == "action_required"
+ assert chain["operator_outcome"]["notification"]["mode"] == "automation_progress"
assert chain["evidence"]["operation_records"] == 1
assert chain["evidence"]["auto_repair_records"] == 0
@@ -2914,7 +2914,7 @@ def test_awooop_status_chain_surfaces_expired_approval_outcome() -> None:
assert chain["repair_state"] == "approval_expired_ai_retry"
assert chain["needs_human"] is False
assert chain["operator_outcome"]["state"] == "approval_expired_ai_retry"
- assert chain["operator_outcome"]["notification"]["mode"] == "action_required"
+ assert chain["operator_outcome"]["notification"]["mode"] == "automation_progress"
def test_legacy_mcp_timeline_summary_surfaces_tool_context() -> None:
diff --git a/apps/api/tests/test_current_owner_controlled_apply_policy.py b/apps/api/tests/test_current_owner_controlled_apply_policy.py
new file mode 100644
index 000000000..f5bbc3e65
--- /dev/null
+++ b/apps/api/tests/test_current_owner_controlled_apply_policy.py
@@ -0,0 +1,236 @@
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pytest
+
+from src.api.v1.webhooks import _controlled_ai_policy_allows
+from src.core.trust_engine import classify_risk, get_required_signatures
+from src.models.approval import ApprovalStatus, RiskLevel
+from src.services import approval_db as approval_db_module
+from src.services.approval_db import ApprovalDBService, approval_request_to_record_data
+from src.services.telegram_gateway import TelegramGateway, TelegramMessage
+
+
+@pytest.mark.parametrize(
+ "risk_level",
+ [RiskLevel.LOW, RiskLevel.MEDIUM, RiskLevel.HIGH],
+)
+def test_noncritical_risks_are_auto_authorized_for_controlled_apply(
+ risk_level: RiskLevel,
+) -> None:
+ assert get_required_signatures(risk_level) == 0
+ assert _controlled_ai_policy_allows(risk_level) is True
+
+
+def test_critical_risk_remains_break_glass() -> None:
+ assert get_required_signatures(RiskLevel.CRITICAL) == 1
+ assert _controlled_ai_policy_allows(RiskLevel.CRITICAL) is False
+
+
+@pytest.mark.parametrize(
+ "action",
+ [
+ "host reboot 192.168.0.110",
+ "kubectl get secret awoooi-secrets",
+ "git force push main",
+ "restore database from snapshot",
+ ],
+)
+def test_critical_action_overrides_explicit_high_risk(action: str) -> None:
+ assert classify_risk(action, explicit_level=RiskLevel.HIGH) == RiskLevel.CRITICAL
+
+
+def test_medium_approval_record_is_created_auto_approved() -> None:
+ request = SimpleNamespace(
+ action="kubectl rollout restart deployment/api",
+ description="bounded rollout",
+ metadata={"source": "alertmanager"},
+ blast_radius=None,
+ dry_run_checks=[],
+ requested_by="OpenClaw",
+ expires_at=None,
+ incident_id="INC-CONTROLLED",
+ matched_playbook_id=None,
+ )
+
+ data = approval_request_to_record_data(
+ request,
+ RiskLevel.MEDIUM,
+ get_required_signatures(RiskLevel.MEDIUM),
+ )
+
+ assert data["status"] == ApprovalStatus.APPROVED
+ assert data["required_signatures"] == 0
+ assert data["resolved_at"] is not None
+
+
+def test_medium_telegram_card_is_automation_progress_not_action_required() -> None:
+ message = TelegramMessage(
+ status_emoji="⚠️",
+ risk_level="MEDIUM",
+ resource_name="awoooi-api",
+ root_cause="restart required",
+ suggested_action="kubectl rollout restart deployment/awoooi-api",
+ estimated_downtime="30s",
+ approval_id="approval-1",
+ incident_id="INC-CONTROLLED",
+ controlled_auto_authorized=True,
+ )
+
+ rendered = message.format()
+
+ assert "AI CONTROLLED AUTOMATION" in rendered
+ assert "ACTION REQUIRED" not in rendered
+ assert "awaiting_approval" not in rendered
+ assert "check>apply>verify>writeback" in rendered
+
+
+def test_critical_telegram_card_stays_break_glass() -> None:
+ message = TelegramMessage(
+ status_emoji="🚨",
+ risk_level="CRITICAL",
+ resource_name="database",
+ root_cause="destructive restore requested",
+ suggested_action="restore database from snapshot",
+ estimated_downtime="unknown",
+ approval_id="approval-critical",
+ incident_id="INC-CRITICAL",
+ )
+
+ rendered = message.format()
+
+ assert "BREAK-GLASS REQUIRED" in rendered
+ assert "AI CONTROLLED AUTOMATION" not in rendered
+
+
+@pytest.mark.asyncio
+async def test_controlled_automation_keyboard_has_no_approval_buttons() -> None:
+ gateway = TelegramGateway()
+
+ keyboard = await gateway._build_inline_keyboard(
+ approval_id="approval-1",
+ incident_id="INC-CONTROLLED",
+ suggested_action="kubectl rollout restart deployment/awoooi-api",
+ controlled_auto_authorized=True,
+ )
+ button_texts = {
+ button["text"]
+ for row in keyboard["inline_keyboard"]
+ for button in row
+ }
+
+ assert "✅ 批准" not in button_texts
+ assert "❌ 拒絕" not in button_texts
+ assert "🧰 處置包" in button_texts
+ assert "📊 歷史" in button_texts
+
+
+def _approval_record(*, action: str, risk_level: RiskLevel) -> SimpleNamespace:
+ now = datetime.now(UTC)
+ return SimpleNamespace(
+ id=str(uuid4()),
+ action=action,
+ description="legacy pending alert",
+ status=ApprovalStatus.PENDING,
+ risk_level=risk_level,
+ required_signatures=1,
+ current_signatures=0,
+ signatures=[],
+ blast_radius={
+ "affected_pods": 1,
+ "estimated_downtime": "30s",
+ "related_services": ["awoooi-api"],
+ "data_impact": "write",
+ },
+ dry_run_checks=[],
+ requested_by="OpenClaw",
+ created_at=now,
+ expires_at=None,
+ resolved_at=None,
+ rejection_reason=None,
+ extra_metadata={"source": "alertmanager"},
+ fingerprint="fingerprint",
+ hit_count=1,
+ last_seen_at=now,
+ incident_id="INC-LEGACY-PENDING",
+ matched_playbook_id=None,
+ telegram_message_id=None,
+ telegram_chat_id=None,
+ )
+
+
+class _FakeResult:
+ def __init__(self, record: SimpleNamespace) -> None:
+ self._record = record
+
+ def scalar_one_or_none(self) -> SimpleNamespace:
+ return self._record
+
+
+class _FakeDB:
+ def __init__(self, record: SimpleNamespace) -> None:
+ self.record = record
+ self.events: list[object] = []
+
+ async def execute(self, _statement: object) -> _FakeResult:
+ return _FakeResult(self.record)
+
+ def add(self, event: object) -> None:
+ self.events.append(event)
+
+ async def flush(self) -> None:
+ return None
+
+ async def refresh(self, _record: object) -> None:
+ return None
+
+
+@pytest.mark.asyncio
+async def test_legacy_medium_pending_gate_is_durably_promoted(monkeypatch) -> None:
+ record = _approval_record(
+ action="kubectl rollout restart deployment/awoooi-api",
+ risk_level=RiskLevel.MEDIUM,
+ )
+ db = _FakeDB(record)
+
+ @asynccontextmanager
+ async def fake_db_context():
+ yield db
+
+ monkeypatch.setattr(approval_db_module, "get_db_context", fake_db_context)
+
+ approval, promoted = await ApprovalDBService().apply_current_owner_policy(record.id)
+
+ assert promoted is True
+ assert approval is not None
+ assert approval.status == ApprovalStatus.APPROVED
+ assert approval.required_signatures == 0
+ assert approval.metadata["legacy_pending_promoted"] is True
+ assert len(db.events) == 1
+
+
+@pytest.mark.asyncio
+async def test_legacy_critical_pending_gate_is_not_promoted(monkeypatch) -> None:
+ record = _approval_record(
+ action="host reboot 192.168.0.110",
+ risk_level=RiskLevel.HIGH,
+ )
+ db = _FakeDB(record)
+
+ @asynccontextmanager
+ async def fake_db_context():
+ yield db
+
+ monkeypatch.setattr(approval_db_module, "get_db_context", fake_db_context)
+
+ approval, promoted = await ApprovalDBService().apply_current_owner_policy(record.id)
+
+ assert promoted is False
+ assert approval is not None
+ assert approval.status == ApprovalStatus.PENDING
+ assert approval.required_signatures == 1
+ assert db.events == []
diff --git a/apps/api/tests/test_operator_outcome.py b/apps/api/tests/test_operator_outcome.py
index 14b7db8a6..4b3a48bac 100644
--- a/apps/api/tests/test_operator_outcome.py
+++ b/apps/api/tests/test_operator_outcome.py
@@ -66,7 +66,7 @@ def test_operator_outcome_marks_diagnostic_only_as_ai_repair_required() -> None:
assert outcome["schema_version"] == "operator_outcome_v1"
assert outcome["state"] == "diagnostic_only_ai_repair_required"
assert outcome["needs_human"] is False
- assert outcome["notification"]["mode"] == "action_required"
+ assert outcome["notification"]["mode"] == "automation_progress"
assert "telegram_sre_war_room" in outcome["notification"]["channels"]
assert outcome["next_action"] == "auto_generate_repair_candidate_from_diagnostic_evidence"
assert outcome["execution_result"]["completion_status"] == (
@@ -199,7 +199,7 @@ def test_operator_outcome_marks_expired_approval_as_manual_review() -> None:
assert outcome["state"] == "approval_expired_ai_retry"
assert outcome["needs_human"] is False
- assert outcome["notification"]["mode"] == "action_required"
+ assert outcome["notification"]["mode"] == "automation_progress"
assert outcome["next_action"] == "ai_retry_or_rebuild_controlled_packet"
assert outcome["execution_result"]["completion_status"] == "expired_route_requeued"
diff --git a/apps/api/tests/test_telegram_integration.py b/apps/api/tests/test_telegram_integration.py
index 8850d3b17..ad7ba800e 100644
--- a/apps/api/tests/test_telegram_integration.py
+++ b/apps/api/tests/test_telegram_integration.py
@@ -131,7 +131,7 @@ class TestTelegramMessageStructure:
result = msg.format()
# 驗證關鍵區塊存在（ADR-075 TYPE-3 新格式）
- assert "ACTION REQUIRED" in result
+ assert "BREAK-GLASS REQUIRED" in result
assert "嚴重" in result # risk_level 改為中文
assert "api-service" in result
assert "INC-20260321-0001" in result