feat(awooop): surface degraded ai route lanes
All checks were successful
CD Pipeline / tests (push) Successful in 1m25s
Code Review / ai-code-review (push) Successful in 12s
CD Pipeline / build-and-deploy (push) Successful in 3m37s
CD Pipeline / post-deploy-checks (push) Successful in 1m44s

This commit is contained in:
Your Name
2026-05-25 13:24:53 +08:00
parent 19d306c720
commit ed3e658578
8 changed files with 455 additions and 17 deletions

View File

@@ -19,6 +19,7 @@ from src.services.ollama_failover_manager import OllamaEndpoint, OllamaRoutingRe
from src.services.ollama_health_monitor import HealthReport, HealthStatus
from src.services.platform_operator_service import (
_ai_route_health_map,
_ai_route_lane_state,
_ai_route_policy_order,
_build_awooop_status_chain,
_callback_reply_event_item,
@@ -1549,12 +1550,87 @@ def test_ai_route_status_response_preserves_route_fields() -> None:
"checked": True,
},
},
"lane_mode": "primary",
"active_lane": {
"provider_name": "ollama_gcp_a",
"health_status": "healthy",
"action_required": False,
},
"skipped_lanes": [],
"operator_action": {
"human_required": False,
"action": "monitor",
"reason": "primary_lane_active",
},
"checked_at": datetime(2026, 5, 19, 12, 0, 0),
})
dumped = response.model_dump(mode="json")
assert dumped["policy_order"][-1]["provider_name"] == "gemini"
assert dumped["selected_provider"] == "ollama_gcp_a"
assert dumped["lane_mode"] == "primary"
def test_ai_route_lane_state_marks_degraded_failover() -> None:
policy = _ai_route_policy_order("deep_rca")
health = {
"ollama_gcp_a": {
"status": "offline",
"reason": "recent_endpoint_failure_cooldown:25s",
},
"ollama_gcp_b": {
"status": "healthy",
"reason": "",
},
"ollama_local": {
"status": "healthy",
"reason": "",
},
}
state = _ai_route_lane_state(
policy_order=policy,
selected_provider="ollama_gcp_b",
health=health,
)
assert state["lane_mode"] == "degraded_failover"
assert state["active_lane"]["provider_name"] == "ollama_gcp_b"
assert len(state["skipped_lanes"]) == 1
assert state["skipped_lanes"][0]["provider_name"] == "ollama_gcp_a"
assert state["skipped_lanes"][0]["role"] == "primary"
assert state["skipped_lanes"][0]["health_status"] == "offline"
assert state["skipped_lanes"][0]["reason"] == "recent_endpoint_failure_cooldown:25s"
assert state["skipped_lanes"][0]["action_required"] is True
assert state["operator_action"] == {
"human_required": True,
"action": "repair_skipped_primary_lane",
"reason": "fallback_lane_active",
}
def test_ai_route_lane_state_marks_cloud_fallback() -> None:
policy = _ai_route_policy_order("deep_rca")
health = {
"ollama_gcp_a": {"status": "offline", "reason": "timeout"},
"ollama_gcp_b": {"status": "offline", "reason": "timeout"},
"ollama_local": {"status": "offline", "reason": "timeout"},
}
state = _ai_route_lane_state(
policy_order=policy,
selected_provider="gemini",
health=health,
)
assert state["lane_mode"] == "cloud_fallback"
assert state["active_lane"]["provider_name"] == "gemini"
assert [lane["provider_name"] for lane in state["skipped_lanes"]] == [
"ollama_gcp_a",
"ollama_gcp_b",
"ollama_local",
]
assert state["operator_action"]["action"] == "restore_ollama_lanes"
@pytest.mark.asyncio