fix(aiops): route backup decisions through ssh

apps/api/tests/test_alertmanager_rule_bypass.py

@@ -4,7 +4,11 @@ from src.api.v1.webhooks import (
     _should_bypass_alertmanager_llm,
     _should_use_alertmanager_rule_first,
 )
-from src.services.decision_manager import _should_escalate_auto_approve_rejection
+from src.services.decision_manager import (
+    _is_host_layer_ssh_category,
+    _is_non_k8s_host_category,
+    _should_escalate_auto_approve_rejection,
+)
 from src.services.telegram_gateway import _format_resolved_guard_stamp
 
 
@@ -84,6 +88,18 @@ def test_manual_gate_reasons_escalate_to_emergency_intervention():
     assert _should_escalate_auto_approve_rejection("critical_operation") is False
 
 
+def test_backup_failure_routes_to_decision_ssh_before_kubectl_parser():
+    assert _is_host_layer_ssh_category("backup_failure") is True
+    assert _is_host_layer_ssh_category("host_resource") is True
+    assert _is_host_layer_ssh_category("kubernetes") is False
+
+
+def test_backup_failure_blocks_k8s_auto_execute():
+    assert _is_non_k8s_host_category("backup_failure") is True
+    assert _is_non_k8s_host_category("host_resource") is True
+    assert _is_non_k8s_host_category("infrastructure") is False
+
+
 def test_resolved_guard_stamp_without_timestamp_is_clean():
     assert _format_resolved_guard_stamp(None) == "✅ 此事件已解決"
 

apps/api/tests/test_callback_dispatcher.py

@@ -21,6 +21,7 @@ from src.services.callback_dispatcher import (
     list_actions_for_category,
     load_action_registry,
     _lookup_context,
+    _resolve_provider_name,
     _resolve_template,
 )
 
@@ -68,6 +69,11 @@ class TestRegistryLoading:
             assert spec and spec.callback_format == "info", \
                 f"{qa} should use info format"
 
+    def test_legacy_provider_aliases_resolve_to_registered_names(self):
+        assert _resolve_provider_name("k8s") == "kubernetes"
+        assert _resolve_provider_name("ssh") == "ssh_host"
+        assert _resolve_provider_name("prometheus") == "prometheus"
+
 
 # =============================================================================
 # Category filtering
@@ -91,6 +97,16 @@ class TestCategoryFiltering:
         assert any(a.callback_format == "info" for a in actions), "需至少 1 個查類"
         assert any(a.callback_format == "nonce" for a in actions), "需至少 1 個寫類"
 
+    def test_backup_failure_has_read_only_diagnostics(self):
+        actions = list_actions_for_category("backup_failure")
+        names = {a.name for a in actions}
+        assert {
+            "backup_check_host_disk",
+            "backup_check_jobs",
+            "backup_check_velero",
+        }.issubset(names)
+        assert all(a.callback_format == "info" for a in actions)
+
 
 # =============================================================================
 # Template variable resolution