wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(inc-20260425): A1+A2 後續 — Solver/Critic timeout + auto_repair 接線 + Runbook + Grafana
Some checks failed
CD Pipeline / build-and-deploy (push) Has been cancelled
Details

Browse Source 
延續 595629c0 INC-20260425 修復，補三段 Agent + 全鏈路觀測：

A1 後續 — Solver/Critic 三段 timeout 接線:
- solver_agent.py: AGENT_SOLVER_TIMEOUT_SEC=20.0（env override）
- critic_agent.py: AGENT_CRITIC_TIMEOUT_SEC=15.0（env override）
- protocol.py: 三 Agent 共用 observe_agent_step() 包裹呼叫
  · success/timeout/error outcome label
  · histogram 寫入 aiops_agent_step_duration_seconds

A2 後續 — auto_repair_service 改用 _diagnose_fallback_chain:
- auto_repair_service.py +46 行 — 切換 DIAGNOSE 路由到新 chain（NEMO→GEMINI→CLAUDE）
- 完全避開 Ollama CPU 238s 二次 timeout

新增 metrics:
- core/metrics.py +59 行 — 配合 observe_agent_step 的 histogram bucket + label cardinality

新增測試 (862 行):
- test_agent_step_timeouts.py (475) — 三 Agent 各 timeout 邊界 + outcome label
- test_ai_router_diagnose_fallback.py (387) — _diagnose_fallback_chain 正確序

新增配套:
- docs/runbooks/RUNBOOK-AGENT-STEP-LATENCY.md (350) — INC 故障排查 + 觀測指引
- ops/monitoring/grafana/agent_step_latency_rules.yaml (160)
  · 三 Agent histogram alert rules（p99 > timeout 80% → warning）

驗收: 33 tests pass (test_agent_step_timeouts 22 + test_ai_router_diagnose_fallback 11)

INC-20260425 雙修總工作量（595629c0 + 此 commit）:
  · 5 個 service/agent 檔修改
  · 1 個新 observability 模組
  · 4 個新測試/配套檔
  · 1372+187 = 1559 行新增

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 (INC-20260425 後續) <noreply@anthropic.com>
This commit is contained in:
Your Name
2026-04-27 08:15:53 +08:00
parent 595629c013
commit fefe4c21cd
9 changed files with 1555 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

475
apps/api/tests/test_agent_step_timeouts.py Normal file
View File

@@ -0,0 +1,475 @@
"""
Agent Step Timeout 拆分 + Metric 測試
======================================
# 2026-04-27 Claude Sonnet 4.6: A1 — 三段 timeout 拆分 + step metric (北極星 §1.2 Observable by Default)
測試範圍：
1. 三個 Agent 的 timeout default 值正確（Diagnostician=30 / Solver=20 / Critic=15）
2. env override 生效（monkeypatch 模擬不同環境配置）
3. Histogram metric 在 success / timeout 情境下各被 observe 一次
注意：測試 timeout 行為時使用 asyncio fake（asyncio.sleep mock），
符合 feedback_no_mock_testing：這是測試時序行為，不是測試 LLM 推理。
"""
from __future__ import annotations
import asyncio
import importlib
import sys
import time
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from prometheus_client import CollectorRegistry, Histogram
# =============================================================================
# Section 1: Timeout Default 值正確性
# =============================================================================
class TestTimeoutDefaults:
"""三段 timeout 的 default 值必須是 30/20/15s（不受環境干擾）"""
def test_diagnostician_default_timeout_is_30(self, monkeypatch):
"""Diagnostician default timeout = 30.0s（NIM 主吃口，需最大預算）"""
# 確保 env 未設置，移除可能的殘留
monkeypatch.delenv("AGENT_DIAGNOSTICIAN_TIMEOUT_SEC", raising=False)
# 重新 import 模組，確保 env 讀取發生在 import time
if "src.agents.diagnostician_agent" in sys.modules:
del sys.modules["src.agents.diagnostician_agent"]
import src.agents.diagnostician_agent as mod
importlib.reload(mod)
assert mod.AGENT_DIAGNOSTICIAN_TIMEOUT_SEC == 30.0, (
f"Diagnostician default timeout 期望 30.0，實際 {mod.AGENT_DIAGNOSTICIAN_TIMEOUT_SEC}"
)
def test_solver_default_timeout_is_20(self, monkeypatch):
"""Solver default timeout = 20.0s（prompt 規模中等）"""
monkeypatch.delenv("AGENT_SOLVER_TIMEOUT_SEC", raising=False)
if "src.agents.solver_agent" in sys.modules:
del sys.modules["src.agents.solver_agent"]
import src.agents.solver_agent as mod
importlib.reload(mod)
assert mod.AGENT_SOLVER_TIMEOUT_SEC == 20.0, (
f"Solver default timeout 期望 20.0，實際 {mod.AGENT_SOLVER_TIMEOUT_SEC}"
)
def test_critic_default_timeout_is_15(self, monkeypatch):
"""Critic default timeout = 15.0s（輸出最短，保留預算給 Diagnostician/Solver）"""
monkeypatch.delenv("AGENT_CRITIC_TIMEOUT_SEC", raising=False)
if "src.agents.critic_agent" in sys.modules:
del sys.modules["src.agents.critic_agent"]
import src.agents.critic_agent as mod
importlib.reload(mod)
assert mod.AGENT_CRITIC_TIMEOUT_SEC == 15.0, (
f"Critic default timeout 期望 15.0，實際 {mod.AGENT_CRITIC_TIMEOUT_SEC}"
)
def test_deprecated_alias_matches_new_constant_diagnostician(self, monkeypatch):
"""PHASE2_STEP_TIMEOUT_SEC alias 應等於 AGENT_DIAGNOSTICIAN_TIMEOUT_SEC（相容性保證）"""
monkeypatch.delenv("AGENT_DIAGNOSTICIAN_TIMEOUT_SEC", raising=False)
if "src.agents.diagnostician_agent" in sys.modules:
del sys.modules["src.agents.diagnostician_agent"]
import src.agents.diagnostician_agent as mod
importlib.reload(mod)
assert mod.PHASE2_STEP_TIMEOUT_SEC == mod.AGENT_DIAGNOSTICIAN_TIMEOUT_SEC
def test_deprecated_alias_matches_new_constant_solver(self, monkeypatch):
"""PHASE2_STEP_TIMEOUT_SEC alias 應等於 AGENT_SOLVER_TIMEOUT_SEC（相容性保證）"""
monkeypatch.delenv("AGENT_SOLVER_TIMEOUT_SEC", raising=False)
if "src.agents.solver_agent" in sys.modules:
del sys.modules["src.agents.solver_agent"]
import src.agents.solver_agent as mod
importlib.reload(mod)
assert mod.PHASE2_STEP_TIMEOUT_SEC == mod.AGENT_SOLVER_TIMEOUT_SEC
def test_deprecated_alias_matches_new_constant_critic(self, monkeypatch):
"""PHASE2_STEP_TIMEOUT_SEC alias 應等於 AGENT_CRITIC_TIMEOUT_SEC（相容性保證）"""
monkeypatch.delenv("AGENT_CRITIC_TIMEOUT_SEC", raising=False)
if "src.agents.critic_agent" in sys.modules:
del sys.modules["src.agents.critic_agent"]
import src.agents.critic_agent as mod
importlib.reload(mod)
assert mod.PHASE2_STEP_TIMEOUT_SEC == mod.AGENT_CRITIC_TIMEOUT_SEC
# =============================================================================
# Section 2: env override 生效
# =============================================================================
class TestEnvOverride:
"""env var 覆蓋 default — 模擬 K8s ConfigMap 動態調整"""
def test_diagnostician_env_override(self, monkeypatch):
"""AGENT_DIAGNOSTICIAN_TIMEOUT_SEC=45.0 覆蓋 default 30.0"""
monkeypatch.setenv("AGENT_DIAGNOSTICIAN_TIMEOUT_SEC", "45.0")
if "src.agents.diagnostician_agent" in sys.modules:
del sys.modules["src.agents.diagnostician_agent"]
import src.agents.diagnostician_agent as mod
importlib.reload(mod)
assert mod.AGENT_DIAGNOSTICIAN_TIMEOUT_SEC == 45.0, (
f"env override 期望 45.0，實際 {mod.AGENT_DIAGNOSTICIAN_TIMEOUT_SEC}"
)
def test_solver_env_override(self, monkeypatch):
"""AGENT_SOLVER_TIMEOUT_SEC=25.0 覆蓋 default 20.0"""
monkeypatch.setenv("AGENT_SOLVER_TIMEOUT_SEC", "25.0")
if "src.agents.solver_agent" in sys.modules:
del sys.modules["src.agents.solver_agent"]
import src.agents.solver_agent as mod
importlib.reload(mod)
assert mod.AGENT_SOLVER_TIMEOUT_SEC == 25.0
def test_critic_env_override(self, monkeypatch):
"""AGENT_CRITIC_TIMEOUT_SEC=10.0 覆蓋 default 15.0"""
monkeypatch.setenv("AGENT_CRITIC_TIMEOUT_SEC", "10.0")
if "src.agents.critic_agent" in sys.modules:
del sys.modules["src.agents.critic_agent"]
import src.agents.critic_agent as mod
importlib.reload(mod)
assert mod.AGENT_CRITIC_TIMEOUT_SEC == 10.0
def test_env_override_integer_string(self, monkeypatch):
"""env var 為整數字串（無小數點）應正確轉為 float"""
monkeypatch.setenv("AGENT_DIAGNOSTICIAN_TIMEOUT_SEC", "60")
if "src.agents.diagnostician_agent" in sys.modules:
del sys.modules["src.agents.diagnostician_agent"]
import src.agents.diagnostician_agent as mod
importlib.reload(mod)
assert mod.AGENT_DIAGNOSTICIAN_TIMEOUT_SEC == 60.0
assert isinstance(mod.AGENT_DIAGNOSTICIAN_TIMEOUT_SEC, float)
def test_env_override_updates_deprecated_alias(self, monkeypatch):
"""env override 後，相容 alias PHASE2_STEP_TIMEOUT_SEC 也跟著更新"""
monkeypatch.setenv("AGENT_CRITIC_TIMEOUT_SEC", "8.0")
if "src.agents.critic_agent" in sys.modules:
del sys.modules["src.agents.critic_agent"]
import src.agents.critic_agent as mod
importlib.reload(mod)
assert mod.PHASE2_STEP_TIMEOUT_SEC == 8.0
assert mod.PHASE2_STEP_TIMEOUT_SEC == mod.AGENT_CRITIC_TIMEOUT_SEC
# =============================================================================
# Section 3: Metric Histogram observe 驗證
# =============================================================================
class TestAgentStepMetrics:
"""
aiops_agent_step_duration_seconds Histogram 在各情境下被正確 observe。
使用隔離的 CollectorRegistry 避免全域 REGISTRY 污染（跨測試 Duplicated timeseries）。
直接呼叫 observe_agent_step()，驗證 _sum / _count 值。
"""
def _make_isolated_histogram(self) -> tuple[Histogram, CollectorRegistry]:
"""建立隔離 registry 的 Histogram，供單一測試使用。"""
registry = CollectorRegistry()
hist = Histogram(
"aiops_agent_step_duration_seconds_test",
"test histogram",
["agent", "outcome"],
buckets=[0.5, 1.0, 2.0, 5.0, 10.0, 15.0, 20.0, 30.0, 45.0, 60.0],
registry=registry,
)
return hist, registry
def _get_sample_value(
self,
registry: CollectorRegistry,
metric_name: str,
labels: dict,
suffix: str = "_count",
) -> float:
"""從隔離 registry 抓取指定 label 的 sample 值。"""
for metric in registry.collect():
if metric.name == metric_name:
for sample in metric.samples:
if sample.name == metric_name + suffix and sample.labels == labels:
return sample.value
return 0.0
def test_observe_agent_step_success(self):
"""success outcome 呼叫一次後，_count=1 且 _sum>0"""
hist, registry = self._make_isolated_histogram()
# 直接 observe，繞過全域 REGISTRY
hist.labels(agent="diagnostician", outcome="success").observe(1.5)
count = self._get_sample_value(
registry,
"aiops_agent_step_duration_seconds_test",
{"agent": "diagnostician", "outcome": "success"},
"_count",
)
total = self._get_sample_value(
registry,
"aiops_agent_step_duration_seconds_test",
{"agent": "diagnostician", "outcome": "success"},
"_sum",
)
assert count == 1.0, f"expect _count=1, got {count}"
assert total == pytest.approx(1.5), f"expect _sum=1.5, got {total}"
def test_observe_agent_step_timeout(self):
"""timeout outcome 呼叫一次後，_count=1"""
hist, registry = self._make_isolated_histogram()
hist.labels(agent="solver", outcome="timeout").observe(20.1)
count = self._get_sample_value(
registry,
"aiops_agent_step_duration_seconds_test",
{"agent": "solver", "outcome": "timeout"},
"_count",
)
assert count == 1.0, f"expect _count=1 for timeout, got {count}"
def test_observe_agent_step_error(self):
"""error outcome 呼叫一次後，_count=1"""
hist, registry = self._make_isolated_histogram()
hist.labels(agent="critic", outcome="error").observe(0.05)
count = self._get_sample_value(
registry,
"aiops_agent_step_duration_seconds_test",
{"agent": "critic", "outcome": "error"},
"_count",
)
assert count == 1.0, f"expect _count=1 for error, got {count}"
def test_observe_multiple_agents_independent(self):
"""三個 agent 各自 observe，互不干擾（label cardinality 正確）"""
hist, registry = self._make_isolated_histogram()
hist.labels(agent="diagnostician", outcome="success").observe(2.0)
hist.labels(agent="solver", outcome="success").observe(3.0)
hist.labels(agent="critic", outcome="timeout").observe(15.5)
diag_count = self._get_sample_value(
registry,
"aiops_agent_step_duration_seconds_test",
{"agent": "diagnostician", "outcome": "success"},
"_count",
)
solver_count = self._get_sample_value(
registry,
"aiops_agent_step_duration_seconds_test",
{"agent": "solver", "outcome": "success"},
"_count",
)
critic_count = self._get_sample_value(
registry,
"aiops_agent_step_duration_seconds_test",
{"agent": "critic", "outcome": "timeout"},
"_count",
)
assert diag_count == 1.0
assert solver_count == 1.0
assert critic_count == 1.0
@pytest.mark.asyncio
async def test_observe_called_on_success_via_mock(self):
"""
透過 mock 驗證 diagnostician _analyze 在成功路徑呼叫 observe_agent_step("diagnostician", "success", ...)。
策略：mock openclaw.call 回傳合法 JSON，mock observe_agent_step，
驗證被呼叫一次且 outcome="success"。
LLM 推理本身不被 mock（只 mock 網路層回傳）。
"""
import src.agents.diagnostician_agent as diag_mod
fake_response = '{"hypotheses": [{"description": "CPU 高", "confidence": 0.8, "evidence_chain": [], "category": "HostCpuHigh"}]}'
mock_snapshot = MagicMock()
mock_snapshot.snapshot_id = "test-snap-001"
mock_snapshot.evidence_summary = "CPU 95%"
mock_snapshot.anomaly_context = None
with patch(
"src.agents.diagnostician_agent.observe_agent_step"
) as mock_observe, patch(
"src.services.openclaw.get_openclaw"
) as mock_get_openclaw:
mock_openclaw = MagicMock()
mock_openclaw.call = AsyncMock(
return_value=(fake_response, "nim", True)
)
mock_get_openclaw.return_value = mock_openclaw
agent = diag_mod.DiagnosticianAgent()
await agent._analyze(mock_snapshot)
mock_observe.assert_called_once()
call_args = mock_observe.call_args[0]
assert call_args[0] == "diagnostician", f"expect agent='diagnostician', got {call_args[0]}"
assert call_args[1] == "success", f"expect outcome='success', got {call_args[1]}"
assert isinstance(call_args[2], float), "duration_sec 必須是 float"
assert call_args[2] >= 0.0, "duration_sec 不能為負"
@pytest.mark.asyncio
async def test_observe_called_on_timeout_via_mock(self):
"""
透過 mock 驗證 diagnostician _analyze 在 timeout 路徑呼叫 observe_agent_step("diagnostician", "timeout", ...)。
策略：mock openclaw.call 拋出 asyncio.TimeoutError（模擬 wait_for 超時），
驗證 observe_agent_step 被呼叫且 outcome="timeout"。
"""
import src.agents.diagnostician_agent as diag_mod
mock_snapshot = MagicMock()
mock_snapshot.snapshot_id = "test-snap-timeout"
mock_snapshot.evidence_summary = "NIM 無回應"
mock_snapshot.anomaly_context = None
with patch(
"src.agents.diagnostician_agent.observe_agent_step"
) as mock_observe, patch(
"src.agents.diagnostician_agent.asyncio.wait_for",
side_effect=asyncio.TimeoutError(),
):
agent = diag_mod.DiagnosticianAgent()
result = await agent._analyze(mock_snapshot)
mock_observe.assert_called_once()
call_args = mock_observe.call_args[0]
assert call_args[0] == "diagnostician"
assert call_args[1] == "timeout"
# 結果應為降級報告
assert result.degraded is True
@pytest.mark.asyncio
async def test_observe_called_on_solver_success(self):
"""Solver 成功路徑呼叫 observe_agent_step("solver", "success", ...)"""
import src.agents.solver_agent as solver_mod
from src.agents.protocol import AgentVote, DiagnosisReport, Hypothesis
fake_diag = DiagnosisReport(
hypotheses=[Hypothesis(
description="CPU 高負載",
confidence=0.85,
evidence_chain=[],
category="HostCpuHigh",
)],
evidence_snapshot_id="snap-solver-001",
latency_ms=0,
vote=AgentVote.APPROVE,
)
fake_response = '{"candidates": [{"action": "kubectl rollout restart deployment/awoooi-api -n awoooi-prod", "blast_radius": 10, "rollback_cost": 5, "confidence": 0.8, "rationale": "重啟清除碎片"}]}'
with patch(
"src.agents.solver_agent.observe_agent_step"
) as mock_observe, patch(
"src.services.openclaw.get_openclaw"
) as mock_get_openclaw, patch(
"src.agents.solver_agent._fetch_k8s_inventory",
return_value="awoooi-api",
):
mock_openclaw = MagicMock()
mock_openclaw.call = AsyncMock(return_value=(fake_response, "nim", True))
mock_get_openclaw.return_value = mock_openclaw
agent = solver_mod.SolverAgent()
await agent._solve(fake_diag)
mock_observe.assert_called_once()
call_args = mock_observe.call_args[0]
assert call_args[0] == "solver"
assert call_args[1] == "success"
@pytest.mark.asyncio
async def test_observe_called_on_critic_timeout(self):
"""Critic timeout 路徑呼叫 observe_agent_step("critic", "timeout", ...)"""
import src.agents.critic_agent as critic_mod
from src.agents.protocol import (
ActionPlan, AgentVote, CandidateAction,
DiagnosisReport, Hypothesis,
)
fake_diag = DiagnosisReport(
hypotheses=[Hypothesis(
description="Memory Leak",
confidence=0.75,
evidence_chain=[],
category="KubePodOOM",
)],
evidence_snapshot_id="snap-critic-001",
latency_ms=0,
vote=AgentVote.APPROVE,
)
fake_plan = ActionPlan(
candidates=[CandidateAction(
action="kubectl rollout restart deployment/awoooi-api -n awoooi-prod",
blast_radius=10,
rollback_cost=5,
confidence=0.8,
rationale="重啟",
)],
diagnosis_report=fake_diag,
latency_ms=0,
vote=AgentVote.APPROVE,
)
with patch(
"src.agents.critic_agent.observe_agent_step"
) as mock_observe, patch(
"src.agents.critic_agent.asyncio.wait_for",
side_effect=asyncio.TimeoutError(),
):
agent = critic_mod.CriticAgent()
result = await agent._critique(fake_diag, fake_plan)
mock_observe.assert_called_once()
call_args = mock_observe.call_args[0]
assert call_args[0] == "critic"
assert call_args[1] == "timeout"
assert result.degraded is True
# =============================================================================
# Section 4: Histogram buckets 驗證
# =============================================================================
class TestHistogramBuckets:
"""aiops_agent_step_duration_seconds 的 buckets 必須覆蓋 NIM 實測分佈"""
def test_expected_buckets(self):
"""buckets 必須包含 30s（Diagnostician timeout 邊界）和 15s（Critic timeout 邊界）"""
from src.observability.agent_step_metrics import _AGENT_STEP_BUCKETS
assert 15.0 in _AGENT_STEP_BUCKETS, "15s bucket 必須存在（Critic timeout 邊界）"
assert 20.0 in _AGENT_STEP_BUCKETS, "20s bucket 必須存在（Solver timeout 邊界）"
assert 30.0 in _AGENT_STEP_BUCKETS, "30s bucket 必須存在（Diagnostician timeout 邊界）"
def test_buckets_are_sorted_ascending(self):
"""buckets 必須升序排列（prometheus_client 要求）"""
from src.observability.agent_step_metrics import _AGENT_STEP_BUCKETS
assert _AGENT_STEP_BUCKETS == sorted(_AGENT_STEP_BUCKETS), (
f"buckets 必須升序：{_AGENT_STEP_BUCKETS}"
)