fix(agent): auto-authorize noncritical alert control

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:

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 == []

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"
 

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