fix(agent): bind pre-decision evidence to runtime
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m37s
CD Pipeline / build-and-deploy (push) Successful in 5m1s
CD Pipeline / post-deploy-checks (push) Successful in 3m35s

This commit is contained in:
ogt
2026-07-10 20:55:09 +08:00
parent 98557b01ed
commit b72b6ee048
11 changed files with 314 additions and 41 deletions

View File

@@ -139,6 +139,38 @@ async def test_audited_provider_strips_internal_audit_context(monkeypatch):
assert audit_calls[0]["incident_id"] == "INC-1"
@pytest.mark.asyncio
async def test_audited_provider_does_not_duplicate_first_class_gateway_audit(monkeypatch):
audit_calls = []
async def fake_record_mcp_call(**kwargs):
audit_calls.append(kwargs)
monkeypatch.setattr(
"src.services.mcp_audit_service.record_mcp_call",
fake_record_mcp_call,
)
provider = FakeProvider()
audited = AuditedMCPToolProvider(provider)
result = await audited.execute(
"kubectl_get",
{
"resource": "pods",
"_mcp_audit": {
"gateway_path": "awooop_mcp_gateway",
"agent_role": "pre_decision_investigator",
"incident_id": "INC-GW-1",
},
},
)
assert result.success is True
assert provider.calls == [("kubectl_get", {"resource": "pods"})]
assert audit_calls == []
@pytest.mark.asyncio
async def test_agent_tool_executor_blocks_disallowed_tool():
restart_tool = _tool("kubernetes", "kubectl_restart")