From b72b6ee048a071e235edba94d088a850c33c8987 Mon Sep 17 00:00:00 2001 From: ogt Date: Fri, 10 Jul 2026 20:55:09 +0800 Subject: [PATCH] fix(agent): bind pre-decision evidence to runtime --- .../awooop_ansible_candidate_backfill_job.py | 107 ++++++++++++++++++ apps/api/src/plugins/mcp/registry.py | 36 +++--- .../services/awooop_ansible_audit_service.py | 14 ++- .../awooop_ansible_check_mode_service.py | 2 +- apps/api/src/services/evidence_snapshot.py | 18 ++- apps/api/src/services/mcp_audit_context.py | 8 ++ .../src/services/pre_decision_investigator.py | 103 ++++++++++++++--- apps/api/tests/test_agent_loop_foundation.py | 32 ++++++ ...t_awooop_ansible_candidate_backfill_job.py | 24 ++++ .../tests/test_awooop_truth_chain_service.py | 1 + .../tests/test_pre_decision_investigator.py | 10 +- 11 files changed, 314 insertions(+), 41 deletions(-) diff --git a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py index a6cf3ba87..01b2cb7d9 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -10,7 +10,9 @@ from __future__ import annotations import asyncio from collections.abc import Awaitable, Callable +from types import SimpleNamespace from typing import Any +from uuid import uuid4 import structlog from sqlalchemy import text @@ -25,10 +27,14 @@ from src.services.awooop_ansible_check_mode_service import ( backfill_missing_auto_repair_execution_receipts_once, run_failed_apply_check_mode_replay_once, ) +from src.services.evidence_snapshot import EvidenceSnapshot +from src.services.pre_decision_investigator import get_pre_decision_investigator logger = structlog.get_logger(__name__) Recorder = Callable[..., Awaitable[bool]] +EvidenceCollector = Callable[..., Awaitable[EvidenceSnapshot | None]] +EvidenceVerifier = Callable[..., Awaitable[bool]] _BACKFILL_DECISION_PATH = "repair_candidate_controlled_queue" _BACKFILL_REASON = ( @@ -60,6 +66,7 @@ async def _fetch_missing_candidate_incidents( resolved_at, verification_result, frequency_snapshot, + affected_services, signals, decision_chain FROM incidents @@ -96,6 +103,72 @@ def _build_backfill_proposal(incident: dict[str, Any]) -> dict[str, Any]: } +def _incident_for_evidence(incident: dict[str, Any]) -> SimpleNamespace: + signals = [ + SimpleNamespace( + alert_name=str(signal.get("alert_name") or ""), + labels=dict(signal.get("labels") or {}), + annotations=dict(signal.get("annotations") or {}), + source=str(signal.get("source") or ""), + ) + for signal in incident.get("signals") or [] + if isinstance(signal, dict) + ] + return SimpleNamespace( + incident_id=str(incident.get("incident_id") or ""), + severity=incident.get("severity"), + alertname=incident.get("alertname"), + affected_services=list(incident.get("affected_services") or []), + signals=signals, + ) + + +async def collect_ansible_candidate_pre_decision_evidence( + *, + incident: dict[str, Any], + project_id: str, + automation_run_id: str, +) -> EvidenceSnapshot | None: + return await get_pre_decision_investigator().investigate( + _incident_for_evidence(incident), + project_id=project_id, + automation_run_id=automation_run_id, + ) + + +async def verify_ansible_candidate_pre_decision_evidence( + *, + snapshot: EvidenceSnapshot | None, + project_id: str, +) -> bool: + if ( + snapshot is None + or snapshot.sensors_succeeded <= 0 + or not isinstance(snapshot.recent_logs, str) + or not snapshot.recent_logs.strip() + ): + return False + async with get_db_context(project_id) as db: + verified = await db.execute( + text(""" + SELECT EXISTS ( + SELECT 1 + FROM incident_evidence + WHERE id = :snapshot_id + AND incident_id = :incident_id + AND post_execution_state IS NULL + AND sensors_succeeded > 0 + AND NULLIF(recent_logs, '') IS NOT NULL + ) + """), + { + "snapshot_id": snapshot.snapshot_id, + "incident_id": snapshot.incident_id, + }, + ) + return verified.scalar() is True + + async def enqueue_missing_ansible_candidates_once( *, project_id: str = "awoooi", @@ -108,6 +181,12 @@ async def enqueue_missing_ansible_candidates_once( retry_replayer: Callable[ ..., Awaitable[dict[str, Any]] ] = run_failed_apply_check_mode_replay_once, + evidence_collector: EvidenceCollector = ( + collect_ansible_candidate_pre_decision_evidence + ), + evidence_verifier: EvidenceVerifier = ( + verify_ansible_candidate_pre_decision_evidence + ), ) -> dict[str, Any]: """Backfill missing Ansible candidate rows for recent unresolved incidents.""" @@ -118,6 +197,8 @@ async def enqueue_missing_ansible_candidates_once( "queued": 0, "already_existing_or_write_skipped": 0, "no_catalog_candidate": 0, + "pre_decision_evidence_ready": 0, + "pre_decision_evidence_blocked": 0, "repair_receipts_backfilled": 0, "failed_apply_retry_replayed": 0, "failed_apply_retry_check_mode_passed": 0, @@ -138,6 +219,8 @@ async def enqueue_missing_ansible_candidates_once( "queued": 0, "already_existing_or_write_skipped": 0, "no_catalog_candidate": 0, + "pre_decision_evidence_ready": 0, + "pre_decision_evidence_blocked": 0, "repair_receipts_backfilled": 0, "failed_apply_retry_replayed": 0, "failed_apply_retry_check_mode_passed": 0, @@ -165,11 +248,35 @@ async def enqueue_missing_ansible_candidates_once( if payload is None: stats["no_catalog_candidate"] += 1 continue + automation_run_id = str(uuid4()) + try: + snapshot = await evidence_collector( + incident=incident, + project_id=project_id, + automation_run_id=automation_run_id, + ) + evidence_ready = await evidence_verifier( + snapshot=snapshot, + project_id=project_id, + ) + except Exception as exc: + logger.warning( + "awooop_ansible_candidate_pre_decision_evidence_failed", + incident_id=incident.get("incident_id"), + automation_run_id=automation_run_id, + error=str(exc), + ) + evidence_ready = False + if not evidence_ready: + stats["pre_decision_evidence_blocked"] += 1 + continue + stats["pre_decision_evidence_ready"] += 1 inserted = await recorder( incident=incident, proposal_data=_build_backfill_proposal(incident), decision_path=_BACKFILL_DECISION_PATH, not_used_reason=_BACKFILL_REASON, + automation_run_id=automation_run_id, ) if inserted: stats["queued"] += 1 diff --git a/apps/api/src/plugins/mcp/registry.py b/apps/api/src/plugins/mcp/registry.py index 6b11dd103..bdb094d56 100644 --- a/apps/api/src/plugins/mcp/registry.py +++ b/apps/api/src/plugins/mcp/registry.py @@ -44,6 +44,10 @@ class AuditedMCPToolProvider(MCPToolProvider): from src.services.mcp_audit_service import monotonic_ms, record_mcp_call audit_context = parameters.get("_mcp_audit") if isinstance(parameters, dict) else None + first_class_gateway_audit = ( + isinstance(audit_context, dict) + and audit_context.get("gateway_path") == "awooop_mcp_gateway" + ) provider_parameters = { key: value for key, value in parameters.items() if key != "_mcp_audit" @@ -54,20 +58,24 @@ class AuditedMCPToolProvider(MCPToolProvider): result = await self.__provider.execute(tool_name, provider_parameters) return result finally: - duration_ms = monotonic_ms() - started - await record_mcp_call( - mcp_server=self.name, - tool_name=tool_name, - input_params=parameters, - output_result=result.output if result else None, - duration_ms=duration_ms, - success=bool(result.success) if result else False, - error_message=result.error if result else "provider_exception", - session_id=audit_context.get("session_id") if isinstance(audit_context, dict) else None, - flywheel_node=audit_context.get("flywheel_node") if isinstance(audit_context, dict) else None, - incident_id=audit_context.get("incident_id") if isinstance(audit_context, dict) else None, - agent_role=audit_context.get("agent_role") if isinstance(audit_context, dict) else None, - ) + # McpGateway writes the canonical audit with its existing DB session. + # Opening a second audit session here can exhaust the pool when the + # investigator fans out several sensors concurrently. + if not first_class_gateway_audit: + duration_ms = monotonic_ms() - started + await record_mcp_call( + mcp_server=self.name, + tool_name=tool_name, + input_params=parameters, + output_result=result.output if result else None, + duration_ms=duration_ms, + success=bool(result.success) if result else False, + error_message=result.error if result else "provider_exception", + session_id=audit_context.get("session_id") if isinstance(audit_context, dict) else None, + flywheel_node=audit_context.get("flywheel_node") if isinstance(audit_context, dict) else None, + incident_id=audit_context.get("incident_id") if isinstance(audit_context, dict) else None, + agent_role=audit_context.get("agent_role") if isinstance(audit_context, dict) else None, + ) async def health_check(self) -> bool: return await self.__provider.health_check() diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index 829cb6d89..208f49924 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/apps/api/src/services/awooop_ansible_audit_service.py @@ -9,6 +9,7 @@ from __future__ import annotations import json from typing import Any +from uuid import UUID, uuid4 import structlog from sqlalchemy import text @@ -525,6 +526,7 @@ def build_ansible_decision_audit_payload( incident_id = str(incident_payload.get("incident_id") or "") input_payload = { "incident_id": incident_id, + "project_id": str(incident_payload.get("project_id") or "awoooi"), "executor": "ansible", "execution_backend": "ansible", "decision_path": decision_path, @@ -585,6 +587,7 @@ async def record_ansible_decision_audit( proposal_data: dict[str, Any], decision_path: str, not_used_reason: str, + automation_run_id: str | None = None, ) -> bool: """Write a best-effort Ansible candidate audit row for one decision.""" @@ -598,7 +601,12 @@ async def record_ansible_decision_audit( return False incident_id = payload["input"]["incident_id"] - project_id = getattr(incident, "project_id", None) or "awoooi" + project_id = str(payload["input"].get("project_id") or "awoooi") + try: + candidate_op_id = str(UUID(automation_run_id)) if automation_run_id else str(uuid4()) + except (TypeError, ValueError): + candidate_op_id = str(uuid4()) + payload["input"]["automation_run_id"] = candidate_op_id try: async with get_db_context(str(project_id)) as db: existing = await db.execute( @@ -617,9 +625,10 @@ async def record_ansible_decision_audit( await db.execute( text(""" INSERT INTO automation_operation_log ( - operation_type, actor, status, incident_id, + op_id, operation_type, actor, status, incident_id, input, output, dry_run_result, tags ) VALUES ( + CAST(:op_id AS uuid), :operation_type, 'decision_manager', :status, @@ -632,6 +641,7 @@ async def record_ansible_decision_audit( """), { "operation_type": payload["operation_type"], + "op_id": candidate_op_id, "status": payload["status"], "incident_id": incident_id, "incident_db_id": _automation_operation_log_incident_id(incident_id), diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index 6d5e4a6ad..253c7f0ae 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -838,7 +838,7 @@ def build_ansible_context_runtime_stage_receipts( mcp_health = _json_loads(evidence.get("mcp_health")) sensors_attempted = max(0, _int_from_value(evidence.get("sensors_attempted"), default=0)) sensors_succeeded = max(0, _int_from_value(evidence.get("sensors_succeeded"), default=0)) - if mcp_health or sensors_attempted > 0: + if any(value is True for value in mcp_health.values()) or sensors_succeeded > 0: receipts.append( _runtime_stage_receipt( claim, diff --git a/apps/api/src/services/evidence_snapshot.py b/apps/api/src/services/evidence_snapshot.py index 0bfaef912..71251a171 100644 --- a/apps/api/src/services/evidence_snapshot.py +++ b/apps/api/src/services/evidence_snapshot.py @@ -71,6 +71,7 @@ class EvidenceSnapshot: """ incident_id: str + project_id: str = "awoooi" # Identifiers snapshot_id: str = field(default_factory=lambda: str(uuid.uuid4())) @@ -206,7 +207,7 @@ class EvidenceSnapshot: self.evidence_summary = self.build_summary() try: - async with get_db_context() as db: + async with get_db_context(self.project_id) as db: record = IncidentEvidence( id=self.snapshot_id, incident_id=self.incident_id, @@ -267,7 +268,7 @@ class EvidenceSnapshot: self.verification_result = verification_result try: - async with get_db_context() as db: + async with get_db_context(self.project_id) as db: stmt_result = await db.execute( update(IncidentEvidence) .where(IncidentEvidence.id == self.snapshot_id) @@ -312,7 +313,7 @@ class EvidenceSnapshot: self.pre_execution_state = pre_state try: - async with get_db_context() as db: + async with get_db_context(self.project_id) as db: stmt_result = await db.execute( update(IncidentEvidence) .where(IncidentEvidence.id == self.snapshot_id) @@ -356,7 +357,7 @@ class EvidenceSnapshot: self.self_healing_detail = detail try: - async with get_db_context() as db: + async with get_db_context(self.project_id) as db: stmt_result = await db.execute( update(IncidentEvidence) .where(IncidentEvidence.id == self.snapshot_id) @@ -386,7 +387,11 @@ class EvidenceSnapshot: raise -async def get_latest_snapshot(incident_id: str) -> EvidenceSnapshot | None: +async def get_latest_snapshot( + incident_id: str, + *, + project_id: str = "awoooi", +) -> EvidenceSnapshot | None: """ 查詢某 Incident 最新的 EvidenceSnapshot（由 snapshot_id 識別）。 @@ -395,7 +400,7 @@ async def get_latest_snapshot(incident_id: str) -> EvidenceSnapshot | None: from sqlalchemy import desc, select try: - async with get_db_context() as db: + async with get_db_context(project_id) as db: result = await db.execute( select(IncidentEvidence) .where(IncidentEvidence.incident_id == incident_id) @@ -409,6 +414,7 @@ async def get_latest_snapshot(incident_id: str) -> EvidenceSnapshot | None: snap = EvidenceSnapshot( incident_id=row.incident_id, + project_id=project_id, snapshot_id=row.id, schema_version=row.schema_version, collected_at=row.collected_at, diff --git a/apps/api/src/services/mcp_audit_context.py b/apps/api/src/services/mcp_audit_context.py index 495bdc5ca..9866b1009 100644 --- a/apps/api/src/services/mcp_audit_context.py +++ b/apps/api/src/services/mcp_audit_context.py @@ -71,6 +71,10 @@ def build_mcp_audit_context( agent_role: str | None = None, gateway_path: str = "legacy_registry_provider", operator_user_id: int | str | None = None, + project_id: str | None = None, + run_id: str | None = None, + trace_id: str | None = None, + agent_id: str | None = None, ) -> dict[str, Any]: """Build the `_mcp_audit` metadata object carried beside tool params.""" @@ -83,6 +87,10 @@ def build_mcp_audit_context( "flywheel_node": flywheel_node, "agent_role": agent_role, "operator_user_id": operator_user_id, + "project_id": project_id, + "run_id": run_id, + "trace_id": trace_id, + "agent_id": agent_id, } context.update({key: value for key, value in optional_values.items() if value is not None}) return context diff --git a/apps/api/src/services/pre_decision_investigator.py b/apps/api/src/services/pre_decision_investigator.py index e7031c34b..d67e8c399 100644 --- a/apps/api/src/services/pre_decision_investigator.py +++ b/apps/api/src/services/pre_decision_investigator.py @@ -29,6 +29,7 @@ import hashlib import json import time from typing import TYPE_CHECKING, Any +from uuid import UUID import structlog @@ -37,7 +38,11 @@ from src.plugins.mcp.gateway import GatewayContext, McpGateway from src.plugins.mcp.registry import AuditedMCPToolProvider from src.services.evidence_snapshot import EvidenceSnapshot from src.services.mcp_audit_context import with_mcp_audit_context -from src.services.mcp_tool_registry import RegisteredTool, SensorDimension, get_mcp_tool_registry +from src.services.mcp_tool_registry import ( + RegisteredTool, + SensorDimension, + get_mcp_tool_registry, +) from src.services.sanitization_service import sanitize, sanitize_dict_values if TYPE_CHECKING: @@ -71,7 +76,13 @@ class PreDecisionInvestigator: def __init__(self) -> None: self._registry = get_mcp_tool_registry() - async def investigate(self, incident: "Incident") -> EvidenceSnapshot: + async def investigate( + self, + incident: "Incident", + *, + project_id: str = "awoooi", + automation_run_id: str | None = None, + ) -> EvidenceSnapshot: """ 主入口：為 Incident 蒐集 8D 感官情報。 @@ -93,8 +104,12 @@ class PreDecisionInvestigator: incident_id = incident.incident_id if hasattr(incident, "incident_id") else str(incident.id) # 1. 計算 fingerprint 並查 cache - fingerprint = _compute_fingerprint(incident) - cached = await _get_cache(fingerprint) + fingerprint = _compute_fingerprint( + incident, + project_id=project_id, + automation_run_id=automation_run_id, + ) + cached = await _get_cache(fingerprint, project_id=project_id) if cached is not None: logger.debug("investigator_cache_hit", incident_id=incident_id, fingerprint=fingerprint) return cached @@ -107,7 +122,10 @@ class PreDecisionInvestigator: incident_labels=labels, ) - snapshot = EvidenceSnapshot(incident_id=incident_id) + snapshot = EvidenceSnapshot( + incident_id=incident_id, + project_id=project_id, + ) snapshot.sensors_attempted = len(tools) # 告警基礎資訊：sensors=0 時 AI 至少知道是什麼告警 @@ -129,7 +147,13 @@ class PreDecisionInvestigator: # 3. 並行蒐集（整體 INVESTIGATOR_TIMEOUT_SEC 保護） try: await asyncio.wait_for( - self._collect_all(snapshot, tools, incident), + self._collect_all( + snapshot, + tools, + incident, + project_id=project_id, + automation_run_id=automation_run_id, + ), timeout=INVESTIGATOR_TIMEOUT_SEC, ) except asyncio.TimeoutError: @@ -241,8 +265,8 @@ class PreDecisionInvestigator: 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 4 8D 升級 """ - from src.services.proactive_inspector import get_proactive_inspector from src.services.log_anomaly_detector import get_log_anomaly_detector + from src.services.proactive_inspector import get_proactive_inspector context: dict[str, Any] = {} @@ -303,12 +327,21 @@ class PreDecisionInvestigator: snapshot: EvidenceSnapshot, tools: list[RegisteredTool], incident: "Incident", + *, + project_id: str = "awoooi", + automation_run_id: str | None = None, ) -> None: """並行呼叫所有工具，結果填入 snapshot。""" params = _build_tool_params(incident) tasks = [ - self._collect_one(snapshot, reg, params) + self._collect_one( + snapshot, + reg, + params, + project_id=project_id, + automation_run_id=automation_run_id, + ) for reg in tools ] await asyncio.gather(*tasks, return_exceptions=True) @@ -318,6 +351,9 @@ class PreDecisionInvestigator: snapshot: EvidenceSnapshot, reg: RegisteredTool, params: dict[str, Any], + *, + project_id: str = "awoooi", + automation_run_id: str | None = None, ) -> None: """執行單一 MCP 工具呼叫，結果填入對應感官維度。""" tool_name = reg.tool.name @@ -333,9 +369,20 @@ class PreDecisionInvestigator: incident_id=snapshot.incident_id, flywheel_node="sense", agent_role="pre_decision_investigator", + project_id=project_id, + run_id=automation_run_id, + trace_id=snapshot.incident_id, + agent_id="pre_decision_investigator", ) result = await asyncio.wait_for( - self._execute_tool(reg, tool_name, audited_params, snapshot.incident_id), + self._execute_tool( + reg, + tool_name, + audited_params, + snapshot.incident_id, + project_id=project_id, + automation_run_id=automation_run_id, + ), timeout=MCP_TOOL_TIMEOUT_SEC, ) @@ -370,6 +417,7 @@ class PreDecisionInvestigator: _duration_ms = int((asyncio.get_event_loop().time() - _started) * 1000) asyncio.create_task(_log_mcp_call_to_timeline( snapshot_incident_id=getattr(snapshot, "incident_id", None), + project_id=project_id, provider_name=reg.provider.name, tool_name=tool_name, status=_mcp_status, @@ -385,6 +433,9 @@ class PreDecisionInvestigator: tool_name: str, audited_params: dict[str, Any], incident_id: str, + *, + project_id: str = "awoooi", + automation_run_id: str | None = None, ): """Route production audited providers through AwoooP MCP Gateway. @@ -395,11 +446,16 @@ class PreDecisionInvestigator: if not isinstance(reg.provider, AuditedMCPToolProvider): return await reg.provider.execute(tool_name, audited_params) - async with get_db_context("awoooi") as db: + try: + run_id = UUID(automation_run_id) if automation_run_id else None + except (TypeError, ValueError): + run_id = None + async with get_db_context(project_id) as db: ctx = GatewayContext( - project_id="awoooi", + project_id=project_id, agent_id="pre_decision_investigator", tool_name=tool_name, + run_id=run_id, trace_id=incident_id, is_shadow=True, environment={"env": "prod"}, @@ -410,6 +466,7 @@ class PreDecisionInvestigator: async def _log_mcp_call_to_timeline( snapshot_incident_id: str | None, + project_id: str, provider_name: str, tool_name: str, status: str, @@ -421,9 +478,11 @@ async def _log_mcp_call_to_timeline( "MCP 呼叫次數/24h > 0" KPI 量測。 """ try: - from sqlalchemy import text as _sql - from src.db.base import get_db_context import json as _json + + from sqlalchemy import text as _sql + + from src.db.base import get_db_context _description = _json.dumps({ "provider": provider_name, "tool": tool_name, @@ -431,7 +490,7 @@ async def _log_mcp_call_to_timeline( "error": error, "duration_ms": duration_ms, }, ensure_ascii=False) - async with get_db_context() as _db: + async with get_db_context(project_id) as _db: await _db.execute( _sql(""" INSERT INTO timeline_events ( @@ -652,19 +711,30 @@ def _build_tool_params(incident: "Incident") -> dict[str, Any]: } -def _compute_fingerprint(incident: "Incident") -> str: +def _compute_fingerprint( + incident: "Incident", + *, + project_id: str = "awoooi", + automation_run_id: str | None = None, +) -> str: """計算 cache key 用的 fingerprint。""" labels = _get_labels(incident) key = ":".join([ + project_id, labels.get("alertname", ""), labels.get("namespace", ""), labels.get("pod", labels.get("name", "")), labels.get("severity", ""), + automation_run_id or "shared-observation", ]) return hashlib.sha256(key.encode()).hexdigest()[:16] -async def _get_cache(fingerprint: str) -> EvidenceSnapshot | None: +async def _get_cache( + fingerprint: str, + *, + project_id: str = "awoooi", +) -> EvidenceSnapshot | None: """從 Redis 取快取的 EvidenceSnapshot（若存在）。""" try: from src.core.redis_client import get_redis @@ -677,6 +747,7 @@ async def _get_cache(fingerprint: str) -> EvidenceSnapshot | None: data = json.loads(raw) snap = EvidenceSnapshot( incident_id=data.get("incident_id", ""), + project_id=project_id, snapshot_id=data.get("snapshot_id", ""), ) snap.evidence_summary = data.get("evidence_summary", "") diff --git a/apps/api/tests/test_agent_loop_foundation.py b/apps/api/tests/test_agent_loop_foundation.py index fe8b58f5a..b1f7442be 100644 --- a/apps/api/tests/test_agent_loop_foundation.py +++ b/apps/api/tests/test_agent_loop_foundation.py @@ -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") diff --git a/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py b/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py index be3bfb2af..48b445f61 100644 --- a/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py +++ b/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py @@ -57,8 +57,26 @@ async def test_backfill_enqueues_catalog_matched_incident(monkeypatch: pytest.Mo yield fake_db recorded: list[dict] = [] + call_order: list[str] = [] + + async def fake_evidence_collector(**kwargs): + call_order.append("evidence") + assert kwargs["project_id"] == "awoooi" + assert kwargs["automation_run_id"] + return MagicMock( + snapshot_id="evidence-1", + incident_id="INC-20260627-NODE110", + sensors_succeeded=1, + recent_logs="sanitized logs", + ) + + async def fake_evidence_verifier(**kwargs): + call_order.append("verify") + assert kwargs["snapshot"].snapshot_id == "evidence-1" + return True async def fake_recorder(**kwargs): + call_order.append("record") recorded.append(kwargs) return True @@ -86,6 +104,8 @@ async def test_backfill_enqueues_catalog_matched_incident(monkeypatch: pytest.Mo recorder=fake_recorder, receipt_backfiller=fake_receipt_backfiller, retry_replayer=fake_retry_replayer, + evidence_collector=fake_evidence_collector, + evidence_verifier=fake_evidence_verifier, ) assert result["queued"] == 1 @@ -95,8 +115,12 @@ async def test_backfill_enqueues_catalog_matched_incident(monkeypatch: pytest.Mo assert result["failed_apply_retry_check_mode_passed"] == 1 assert result["failed_apply_retry_check_mode_failed"] == 0 assert result["failed_apply_retry_stage_receipt_written"] == 1 + assert result["pre_decision_evidence_ready"] == 1 + assert result["pre_decision_evidence_blocked"] == 0 + assert call_order == ["evidence", "verify", "record"] assert recorded[0]["decision_path"] == "repair_candidate_controlled_queue" assert recorded[0]["incident"]["incident_id"] == "INC-20260627-NODE110" + assert recorded[0]["automation_run_id"] @pytest.mark.asyncio diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index 910713e31..c278d142b 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -1368,6 +1368,7 @@ def test_ansible_decision_audit_payload_marks_repair_candidate_queue_claimable() ) assert payload["input"]["executor_candidates"] assert payload["dry_run_result"]["check_mode_executed"] is False + assert payload["input"]["project_id"] == "awoooi" def test_ansible_decision_audit_payload_exposes_check_mode_safety_flags() -> None: diff --git a/apps/api/tests/test_pre_decision_investigator.py b/apps/api/tests/test_pre_decision_investigator.py index 0097996ed..22a44cf79 100644 --- a/apps/api/tests/test_pre_decision_investigator.py +++ b/apps/api/tests/test_pre_decision_investigator.py @@ -21,6 +21,7 @@ from __future__ import annotations import asyncio from typing import Any + import pytest from src.plugins.mcp.interfaces import MCPTool, MCPToolProvider, MCPToolResult @@ -33,12 +34,11 @@ from src.services.mcp_tool_registry import ( ) from src.services.pre_decision_investigator import ( PreDecisionInvestigator, + _build_tool_params, _compute_fingerprint, _fill_snapshot_dimension, - _build_tool_params, ) - # ───────────────────────────────────────────────────────────────────────────── # Stubs # ───────────────────────────────────────────────────────────────────────────── @@ -207,6 +207,12 @@ class TestComputeFingerprint: i2 = _stub_incident("Host", "prod", "pod1", "critical") assert _compute_fingerprint(i1) != _compute_fingerprint(i2) + def test_automation_run_id_prevents_cross_run_cache_reuse(self): + incident = _stub_incident("HostErrorLogFlood", "infra", "", "warning") + fp1 = _compute_fingerprint(incident, automation_run_id="run-1") + fp2 = _compute_fingerprint(incident, automation_run_id="run-2") + assert fp1 != fp2 + def test_fingerprint_length_16(self): i = _stub_incident() fp = _compute_fingerprint(i)