feat(aiops): enable read-only agent loop canary
All checks were successful
CD Pipeline / tests (push) Successful in 1m43s
Code Review / ai-code-review (push) Successful in 31s
CD Pipeline / build-and-deploy (push) Successful in 10m22s
CD Pipeline / post-deploy-checks (push) Successful in 4m3s

This commit is contained in:
Your Name
2026-05-01 14:20:16 +08:00
parent 33a7148916
commit f8e44971c1
9 changed files with 315 additions and 8 deletions

View File

@@ -2,6 +2,7 @@ import pytest
from src.plugins.mcp.interfaces import MCPTool, MCPToolProvider, MCPToolResult
from src.plugins.mcp.registry import AuditedMCPToolProvider
from src.services.ai_providers.interfaces import AIResult
from src.services.ai_providers.agent_loop import AgentToolExecutor
from src.services.ai_providers.permissions import filter_tools_for_agent, is_tool_allowed
from src.services.ai_providers.tool_schema import (
@@ -119,3 +120,72 @@ async def test_agent_tool_executor_blocks_disallowed_tool():
assert result.success is False
assert "not allowed" in (result.error or "")
assert provider.calls == []
@pytest.mark.asyncio
async def test_openclaw_agent_loop_shadow_uses_read_only_tools(monkeypatch):
from src.core.config import settings
from src.services.openclaw import OpenClawService
monkeypatch.setattr(settings, "ENABLE_OPENCLAW_AGENT_LOOP_SHADOW", True)
monkeypatch.setattr(settings, "OPENCLAW_AGENT_LOOP_MAX_ITERATIONS", 2)
class FakeAIProvider:
def __init__(self):
self.seen_tools = []
async def analyze_with_tools(
self,
prompt,
available_tools,
tool_executor,
max_iterations=5,
agent_role="openclaw",
context=None,
):
self.seen_tools = available_tools
return AIResult(
raw_response='{"root_cause_check":"ok"}',
success=True,
provider="ollama_agent_loop",
latency_ms=12.3,
)
class FakeAIRegistry:
def __init__(self, provider):
self.provider = provider
def get(self, name):
return self.provider if name == "ollama" else None
class FakeMCPRegistry:
def all(self):
return [FakeProvider("database"), FakeProvider("ssh_host")]
async def fake_list_tools(self):
if self.name == "database":
return [
_tool("database", "list_incidents"),
_tool("database", "execute_sql"),
]
return [_tool("ssh_host", "run_command")]
fake_ai_provider = FakeAIProvider()
monkeypatch.setattr(FakeProvider, "list_tools", fake_list_tools)
monkeypatch.setattr("src.services.ai_router.get_ai_registry", lambda: FakeAIRegistry(fake_ai_provider))
monkeypatch.setattr("src.plugins.mcp.registry.get_provider_registry", lambda: FakeMCPRegistry())
service = object.__new__(OpenClawService)
proposal = {"risk_level": "medium", "action": "investigate"}
await service._maybe_run_openclaw_agent_loop_shadow(
proposal=proposal,
incident_id="INC-1",
severity="P2",
signals=[{"alertname": "ApiErrorRateHigh"}],
affected_services=["awoooi-api"],
expert_context={"source": "test"},
)
assert proposal["agent_loop_shadow"]["success"] is True
assert [tool.name for tool in fake_ai_provider.seen_tools] == ["list_incidents"]