fix(aiops): fallback and escalate automation blockers

apps/api/models.json

@@ -17,8 +17,8 @@
       "endpoint": "http://192.168.0.111:11434",
       "api_path": "/api/generate",
       "models": {
-        "default": "deepseek-r1:14b",
-        "rca": "deepseek-r1:14b",
+        "default": "qwen2.5:7b-instruct",
+        "rca": "qwen2.5:7b-instruct",
         "summary": "gemma3:4b",
         "drift_summary": "qwen2.5:7b-instruct",
         "drift_intent": "qwen2.5:7b-instruct",

apps/api/src/api/v1/drift.py

@@ -180,6 +180,7 @@ async def _analyze_and_notify(report: DriftReport) -> None:
 
         # 2026-04-24: 嘗試低風險自動採納
         auto_adopted = False
+        auto_block_reason = ""
         try:
             adopt_svc = get_drift_adopt_service()
             auto_result = await adopt_svc.auto_adopt_if_safe(report)
@@ -197,19 +198,28 @@ async def _analyze_and_notify(report: DriftReport) -> None:
                     pr_url=auto_result.get("pr_url"),
                 )
             else:
+                auto_block_reason = auto_result.get("reason", "") or "auto adopt skipped"
                 _logger.info(
                     "drift_auto_adopt_skipped",
                     report_id=report.report_id,
-                    reason=auto_result.get("reason", ""),
+                    reason=auto_block_reason,
                     skipped=auto_result.get("skipped", True),
                 )
         except Exception as e:
+            auto_block_reason = f"auto adopt error: {str(e)[:120]}"
             _logger.warning("drift_auto_adopt_error", report_id=report.report_id, error=str(e))
 
         if auto_adopted:
             # 自動採納成功，Telegram 通知已在 auto_adopt_if_safe 內發出，不再推送按鈕卡片
             return
 
+        if auto_block_reason:
+            await _escalate_drift_auto_adopt_blocked(
+                report=report,
+                reason=auto_block_reason,
+                interpretation=interpretation,
+            )
+
         # ADR-075: drift_narrator_service 負責發送 TYPE-4D 卡片（含按鈕）
         # 舊的 send_text() 已移除，改由 narrate_and_notify() 統一處理
         try:
@@ -224,6 +234,25 @@ async def _analyze_and_notify(report: DriftReport) -> None:
         structlog.get_logger(__name__).error("drift_analyze_notify_failed", error=str(e))
 
 
+async def _escalate_drift_auto_adopt_blocked(
+    *,
+    report: DriftReport,
+    reason: str,
+    interpretation,
+) -> None:
+    """Delegate drift emergency escalation to the service layer."""
+
+    from src.services.emergency_escalation_service import (
+        escalate_drift_auto_adopt_blocked,
+    )
+
+    await escalate_drift_auto_adopt_blocked(
+        report=report,
+        reason=reason,
+        interpretation=interpretation,
+    )
+
+
 async def _run_full_scan(namespaces: list[str]) -> None:
     """背景：完整漂移掃描"""
     detector = get_drift_detector()

apps/api/src/api/v1/webhooks.py

@@ -110,6 +110,34 @@ def _should_bypass_alertmanager_llm(
         and alert_category == "host_resource"
     )
 
+
+async def _escalate_auto_repair_unavailable(
+    *,
+    incident_id: str,
+    approval_id: str,
+    alert_type: str,
+    target_resource: str,
+    namespace: str,
+    failure_reason: str,
+    attempted_actions: str,
+) -> None:
+    """Delegate automation-blocker escalation to the service layer."""
+
+    from src.services.emergency_escalation_service import (
+        escalate_auto_repair_unavailable,
+    )
+
+    await escalate_auto_repair_unavailable(
+        incident_id=incident_id,
+        approval_id=approval_id,
+        alert_type=alert_type,
+        target_resource=target_resource,
+        namespace=namespace,
+        failure_reason=failure_reason,
+        attempted_actions=attempted_actions,
+    )
+
+
 async def _try_auto_repair_background(
     incident_id: str,
     approval_id: str,
@@ -165,6 +193,15 @@ async def _try_auto_repair_background(
                     "playbook_id": decision.playbook.playbook_id if decision.playbook else None,
                 },
             )
+            await _escalate_auto_repair_unavailable(
+                incident_id=incident_id,
+                approval_id=approval_id,
+                alert_type=alert_type,
+                target_resource=target_resource,
+                namespace=namespace,
+                failure_reason=decision.reason,
+                attempted_actions=f"evaluate_auto_repair -> blocked:{decision.blocked_by}",
+            )
             return
 
         # 記錄自動修復觸發 (Sprint 5.1 Q10: 加入 Langfuse trace_id 追蹤)
@@ -228,6 +265,16 @@ async def _try_auto_repair_background(
                     "namespace": namespace,
                 },
             )
+            if not result.success:
+                await _escalate_auto_repair_unavailable(
+                    incident_id=incident_id,
+                    approval_id=approval_id,
+                    alert_type=alert_type,
+                    target_resource=target_resource,
+                    namespace=namespace,
+                    failure_reason=result.error or "auto repair execution failed",
+                    attempted_actions=f"execute_auto_repair playbook:{result.playbook_id}",
+                )
 
         # ADR-073 Phase 2-3: 自動修復結果 → 寫入 Incident outcome (2026-04-12 ogt)
         # 讓 KMConversionService 可依 outcome 判斷是否為 EXECUTION_SUCCESS

apps/api/src/core/config.py

@@ -525,7 +525,7 @@ class Settings(BaseSettings):
         description="OpenClaw AI Agent service URL",
     )
     OPENCLAW_DEFAULT_MODEL: str = Field(
-        default="deepseek-r1:14b",  # 2026-04-08 ogt: SRE最強推理，M1 Pro實測 13 tok/s
+        default="qwen2.5:7b-instruct",  # 2026-04-30: DIAGNOSE/RCA 優先低延遲本地模型
         description="Default Ollama model for RCA analysis",
     )
     OPENCLAW_TIMEOUT: int = Field(

apps/api/src/services/decision_manager.py

@@ -54,6 +54,26 @@ def _fire_and_forget(coro) -> asyncio.Task:
     return task
 
 
+def _phase2_fallback_reason(package: Any) -> str | None:
+    """Return why a Phase 2 package should continue to Playbook/LLM fallback.
+
+    Phase 2 is a deliberation layer, not the final brake. If the debate times
+    out or produces no actionable recommendation, the flywheel must still try
+    deterministic Playbook/LLM/expert paths before opening a manual gate.
+    """
+
+    status = getattr(package, "session_status", None)
+    status_value = status.value if status and hasattr(status, "value") else str(status or "")
+    if status_value in {"timeout", "failed"}:
+        return f"session_{status_value}"
+    if getattr(package, "all_agents_degraded", False):
+        return "all_agents_degraded"
+    action = str(getattr(package, "recommended_action", "") or "").strip()
+    if not action and getattr(package, "requires_human_approval", False):
+        return "empty_action_requires_human"
+    return None
+
+
 # =============================================================================
 # Phase 31 (ADR-067 2026-04-10): Log 異常摘要 — NemoTron deepseek-r1:14b
 # =============================================================================
@@ -2455,13 +2475,26 @@ class DecisionManager:
                 # evidence_snapshot 攜帶進 proposal_data，避免 singleton 並發污染
                 _p2_result = _package_to_proposal_data(package)
                 _p2_result["_evidence_snapshot_ref"] = p2_snapshot
-                return _p2_result
+                _p2_fallback = _phase2_fallback_reason(package)
+                if not _p2_fallback:
+                    return _p2_result
+
+                logger.warning(
+                    "p2_degraded_fallback_to_playbook_llm",
+                    incident_id=incident.incident_id,
+                    reason=_p2_fallback,
+                    confidence=_p2_result.get("confidence"),
+                    blocked_reason=_p2_result.get("blocked_reason", ""),
+                )
             # snapshot 仍為 None → 降級繼續走原路徑（不阻塞）
-            logger.warning("p2_no_snapshot_fallback", incident_id=incident.incident_id)
+            if p2_snapshot is None:
+                logger.warning("p2_no_snapshot_fallback", incident_id=incident.incident_id)
 
         # Phase 7.5: 先嘗試 Playbook 匹配
         playbook_result = await self._try_playbook_match(incident)
         if playbook_result:
+            if evidence_snapshot is not None:
+                playbook_result["_evidence_snapshot_ref"] = evidence_snapshot
             return playbook_result
 
         # MCP Phase 4c: Playbook 無命中 → 非同步產生 AI 草稿 Playbook (2026-04-11 Claude Sonnet 4.6)

apps/api/src/services/emergency_escalation_service.py

@@ -0,0 +1,154 @@
+"""Emergency escalation service for automation blockers.
+
+Keeps Redis dedup, Telegram fanout, and operation-log writes out of API
+routers while giving auto-repair / drift paths a single emergency channel.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import structlog
+
+from src.core.config import settings
+from src.core.redis_client import get_redis
+
+logger = structlog.get_logger(__name__)
+
+
+async def escalate_auto_repair_unavailable(
+    *,
+    incident_id: str,
+    approval_id: str,
+    alert_type: str,
+    target_resource: str,
+    namespace: str,
+    failure_reason: str,
+    attempted_actions: str,
+) -> None:
+    """Open an emergency channel when auto repair cannot safely continue."""
+
+    dedup_key = f"auto_repair:emergency_escalated:{incident_id}"
+    if not await _dedup_first_send(dedup_key, ttl=900, event="auto_repair"):
+        logger.info(
+            "auto_repair_escalation_dedup_skipped",
+            incident_id=incident_id,
+            approval_id=approval_id,
+        )
+        return
+
+    try:
+        from src.repositories.alert_operation_log_repository import (
+            get_alert_operation_log_repository,
+        )
+        from src.services.telegram_gateway import get_telegram_gateway
+
+        await get_telegram_gateway().send_escalation_card(
+            incident_id=incident_id,
+            original_alertname=alert_type or "AutoRepairBlocked",
+            duration_min=0,
+            priority=0,
+            attempted_actions=attempted_actions,
+            failure_reason=failure_reason,
+            current_impact=f"target={target_resource or 'unknown'} namespace={namespace or 'unknown'}",
+            group_chat_id=settings.SRE_GROUP_CHAT_ID or None,
+        )
+
+        await get_alert_operation_log_repository().append(
+            "EMERGENCY_ESCALATED",
+            incident_id=incident_id,
+            approval_id=approval_id,
+            actor="auto_repair",
+            action_detail="auto_repair_unavailable_emergency_channel",
+            success=True,
+            context={
+                "alert_type": alert_type,
+                "target_resource": target_resource,
+                "namespace": namespace,
+                "failure_reason": failure_reason,
+                "attempted_actions": attempted_actions,
+            },
+        )
+        logger.warning(
+            "auto_repair_emergency_escalated",
+            incident_id=incident_id,
+            approval_id=approval_id,
+            reason=failure_reason,
+        )
+    except Exception as exc:
+        logger.warning(
+            "auto_repair_emergency_escalation_failed",
+            incident_id=incident_id,
+            approval_id=approval_id,
+            error=str(exc),
+        )
+
+
+async def escalate_drift_auto_adopt_blocked(
+    *,
+    report: Any,
+    reason: str,
+    interpretation: Any,
+) -> None:
+    """Notify the emergency channel when drift cannot be auto-adopted safely."""
+
+    dedup_key = f"drift:auto_adopt_emergency:{report.report_id}"
+    if not await _dedup_first_send(dedup_key, ttl=3600, event="drift"):
+        logger.info("drift_emergency_escalation_dedup_skipped", report_id=report.report_id)
+        return
+
+    try:
+        from src.services.telegram_gateway import get_telegram_gateway
+
+        actionable_count = sum(
+            1 for item in report.items
+            if not getattr(item, "is_allowlisted", False)
+        )
+        intent = getattr(getattr(interpretation, "intent", None), "value", "unknown")
+        confidence = getattr(interpretation, "confidence", 0.0) if interpretation else 0.0
+        risk = getattr(interpretation, "risk", "unknown") if interpretation else "unknown"
+
+        await get_telegram_gateway().send_escalation_card(
+            incident_id=report.report_id,
+            original_alertname="ConfigDriftAutoAdoptBlocked",
+            duration_min=0,
+            priority=0 if report.high_count > 0 else 1,
+            attempted_actions="drift_interpreter -> auto_adopt_if_safe -> emergency_escalation",
+            failure_reason=reason,
+            current_impact=(
+                f"namespace={report.namespace} high={report.high_count} "
+                f"medium={report.medium_count} actionable={actionable_count} "
+                f"intent={intent} confidence={confidence:.0%} risk={risk}"
+            ),
+            group_chat_id=settings.SRE_GROUP_CHAT_ID or None,
+        )
+        logger.warning(
+            "drift_auto_adopt_emergency_escalated",
+            report_id=report.report_id,
+            reason=reason,
+            high=report.high_count,
+            medium=report.medium_count,
+            actionable=actionable_count,
+        )
+    except Exception as exc:
+        logger.warning(
+            "drift_emergency_escalation_failed",
+            report_id=report.report_id,
+            error=str(exc),
+        )
+
+
+async def _dedup_first_send(key: str, *, ttl: int, event: str) -> bool:
+    """Return True when this is the first escalation in the dedup window."""
+
+    try:
+        redis = get_redis()
+        return bool(await redis.set(key, "1", ex=ttl, nx=True))
+    except Exception as exc:
+        logger.warning(
+            "emergency_escalation_dedup_failed_open",
+            key=key,
+            event=event,
+            error=str(exc),
+        )
+        return True

apps/api/src/services/failover_alerter.py

@@ -131,9 +131,9 @@ class FailoverAlerter:
 
         msg = (
             f"*Gemini 每日配額耗盡*\n\n"
-            f"日期：{date_str}\n"
-            f"上限：{quota} calls/day\n"
-            f"當前用量：{current_count}\n"
+            f"日期：{_escape_md(date_str)}\n"
+            f"上限：{_escape_md(str(quota))} calls/day\n"
+            f"當前用量：{_escape_md(str(current_count))}\n"
             f"降級目標：Nemotron → Claude \\(Gemini 不可用\\)\n\n"
             f"進入容災模式至明日 0:00\n"
             f"建議檢查是否有異常流量，評估是否升級 Gemini 配額"

apps/api/tests/test_failover_alerter.py

@@ -145,6 +145,7 @@ async def test_quota_alert_dedup_24h(mock_redis, mock_telegram_send):
     assert "Gemini 每日配額耗盡" in sent
     assert "1000" in sent
     assert "1003" in sent
+    assert "\\-" in sent  # MarkdownV2 date hyphens must be escaped
 
     # 驗證 dedup TTL = 24h
     assert mock_redis.set.await_args_list[0].kwargs["ex"] == QUOTA_DEDUP_TTL_SEC

apps/api/tests/test_phase2_fallback.py

@@ -0,0 +1,37 @@
+from types import SimpleNamespace
+
+from src.agents.protocol import AgentSessionStatus
+from src.services.decision_manager import _phase2_fallback_reason
+
+
+def _package(**kwargs):
+    data = {
+        "session_status": AgentSessionStatus.COMPLETED,
+        "all_agents_degraded": False,
+        "recommended_action": "kubectl rollout restart deployment/awoooi-api -n awoooi-prod",
+        "requires_human_approval": False,
+    }
+    data.update(kwargs)
+    return SimpleNamespace(**data)
+
+
+def test_phase2_timeout_continues_to_fallback() -> None:
+    pkg = _package(
+        session_status=AgentSessionStatus.TIMEOUT,
+        recommended_action=None,
+        requires_human_approval=True,
+    )
+
+    assert _phase2_fallback_reason(pkg) == "session_timeout"
+
+
+def test_phase2_empty_manual_gate_continues_to_fallback() -> None:
+    pkg = _package(recommended_action="", requires_human_approval=True)
+
+    assert _phase2_fallback_reason(pkg) == "empty_action_requires_human"
+
+
+def test_phase2_actionable_package_stays_primary() -> None:
+    pkg = _package()
+
+    assert _phase2_fallback_reason(pkg) is None