From 1fa336637233b949dc431ed991156d495bb2ce31 Mon Sep 17 00:00:00 2001 From: matt butler Date: Thu, 9 Jul 2026 14:33:38 -0400 Subject: [PATCH] feat(agent): enforce unresolved-conflict escalation to the coordinator Closes #82. ConflictRecord defaults resolution="unresolved" and the schema states sub-agents must never silently resolve a conflict; they hand it to the coordinator. But escalate() routed purely by report.severity and never looked at diagnosis.conflicts, so a diagnosis carrying contradictory evidence could pass at LOW/MEDIUM and be logged-and-forgotten. Acting autonomously on contradictory evidence is the exact failure mode the contract exists to stop. escalate() now, whenever the diagnosis has unresolved conflicts: - surfaces them first (ERROR log + prominent printout), before any severity prompt, so the audit trail never shows a silent pass; - forces a human acknowledgment gate for non-CRITICAL incidents (CRITICAL already gates, and the conflicts are surfaced before that prompt, so no double-prompt); - records unresolved_conflict_count and conflicts_acknowledged in the audit entry. Degrades gracefully with no stdin (automated run): the conflict is still surfaced and audited, acknowledgment recorded as None, never dropped. Tests (6): surfaced + gated + audited on accept/reject/no-stdin, resolved conflicts not gated, no-conflict path unchanged, and CRITICAL+conflict prompts once not twice. Full suite 222 passed; ruff + mypy clean. --- mcp-server/src/streamops_mcp/agent/audit.py | 7 + .../src/streamops_mcp/agent/escalation.py | 103 +++++++++- mcp-server/tests/test_escalation.py | 194 +++++++++++++++++- 3 files changed, 300 insertions(+), 4 deletions(-) diff --git a/mcp-server/src/streamops_mcp/agent/audit.py b/mcp-server/src/streamops_mcp/agent/audit.py index 1e6d8cf..5bb9c20 100644 --- a/mcp-server/src/streamops_mcp/agent/audit.py +++ b/mcp-server/src/streamops_mcp/agent/audit.py @@ -32,6 +32,8 @@ def log_incident( report: IncidentReport, diagnosis: DiagnosisReport | None = None, human_approved: bool | None = None, + unresolved_conflict_count: int = 0, + conflicts_acknowledged: bool | None = None, ) -> dict: entry: dict[str, Any] = { "timestamp": datetime.now(UTC).isoformat(), @@ -46,6 +48,11 @@ def log_incident( "low_confidence_claims": report.low_confidence_claims, "requires_human_approval": report.requires_human_approval, "human_approved": human_approved, + # Non-zero means the diagnosis had contradictions the coordinator was + # forced to surface; conflicts_acknowledged records the human decision + # (None = no input / already gated by CRITICAL severity). + "unresolved_conflict_count": unresolved_conflict_count, + "conflicts_acknowledged": conflicts_acknowledged, } if diagnosis is not None: diff --git a/mcp-server/src/streamops_mcp/agent/escalation.py b/mcp-server/src/streamops_mcp/agent/escalation.py index 5aa905b..4b818be 100644 --- a/mcp-server/src/streamops_mcp/agent/escalation.py +++ b/mcp-server/src/streamops_mcp/agent/escalation.py @@ -9,17 +9,42 @@ CRITICAL incidents require a human to acknowledge before the system takes any remediation action, preventing autonomous systems from making things worse during a severe outage. + +Independently of severity, a diagnosis carrying UNRESOLVED conflicts is never +allowed to pass silently. The ConflictRecord contract says sub-agents must +never resolve a conflict themselves; they mark it unresolved and hand it to +the coordinator. Acting autonomously on contradictory evidence is exactly the +failure mode to prevent, so unresolved conflicts are always surfaced and, +unless severity already forced a human decision, gated on human acknowledgment. """ import logging from streamops_mcp.agent.audit import AuditLogger -from streamops_mcp.agent.schemas import DiagnosisReport, IncidentReport, Severity +from streamops_mcp.agent.schemas import ( + ConflictRecord, + DiagnosisReport, + IncidentReport, + Severity, +) logger = logging.getLogger("streamops-mcp.escalation") _audit = AuditLogger() +UNRESOLVED = "unresolved" + + +def _unresolved_conflicts(diagnosis: DiagnosisReport | None) -> list[ConflictRecord]: + """Return the conflicts the diagnostic sub-agent left for the coordinator. + + Per the ConflictRecord contract these are marked ``resolution="unresolved"``; + they are the ones the escalation path must act on rather than drop. + """ + if diagnosis is None: + return [] + return [c for c in diagnosis.conflicts if c.resolution == UNRESOLVED] + async def escalate( report: IncidentReport, @@ -28,6 +53,12 @@ async def escalate( """Route an incident report through the escalation chain.""" logger.info("Escalating incident '%s' (severity=%s)", report.title, report.severity.value) + unresolved = _unresolved_conflicts(diagnosis) + if unresolved: + # Always surface first, so the human sees the contradictions before any + # severity prompt and the audit trail never shows a silent pass. + _surface_unresolved_conflicts(report, unresolved) + human_approved = None if report.severity == Severity.LOW: @@ -39,7 +70,19 @@ async def escalate( elif report.severity == Severity.CRITICAL: human_approved = await _handle_critical(report) - _audit.log_incident(report, diagnosis=diagnosis, human_approved=human_approved) + # CRITICAL already forced a human decision (with the conflicts surfaced above); + # for anything lower, unresolved conflicts force their own acknowledgment gate. + conflicts_acknowledged = None + if unresolved and report.severity != Severity.CRITICAL: + conflicts_acknowledged = await _confirm_unresolved_conflicts(report, unresolved) + + _audit.log_incident( + report, + diagnosis=diagnosis, + human_approved=human_approved, + unresolved_conflict_count=len(unresolved), + conflicts_acknowledged=conflicts_acknowledged, + ) async def _handle_low(report: IncidentReport) -> None: @@ -112,3 +155,59 @@ def _print_recommended_actions(report: IncidentReport) -> None: print("Recommended actions:") for i, action in enumerate(report.recommended_actions, 1): print(f" {i}. {action.action} (risk: {action.risk})") + + +def _surface_unresolved_conflicts(report: IncidentReport, conflicts: list[ConflictRecord]) -> None: + """Log and print unresolved conflicts prominently so none passes silently.""" + logger.error( + "Incident '%s' carries %d UNRESOLVED diagnostic conflict(s); forcing review", + report.title, + len(conflicts), + ) + print("\n" + "!" * 60) + print(f"UNRESOLVED DIAGNOSTIC CONFLICTS: {len(conflicts)}") + print("!" * 60) + for c in conflicts: + print(f" - [{c.conflict_id}] {c.topic}: claim {c.claim_a_id} vs {c.claim_b_id}") + if c.notes: + print(f" notes: {c.notes}") + + +async def _confirm_unresolved_conflicts( + report: IncidentReport, conflicts: list[ConflictRecord] +) -> bool | None: + """Force a human decision on a diagnosis with unresolved conflicts. + + Returns True if a human accepted the diagnosis despite the conflicts, False + if rejected, None if no input is available (automated run). In every case + the conflicts have already been surfaced and are recorded in the audit, so + the diagnosis never passes silently. + """ + print( + f"\nDiagnosis for '{report.title}' has {len(conflicts)} unresolved " + "conflict(s). Accept for escalation anyway? (y/n): ", + end="", + flush=True, + ) + try: + response = input() + except EOFError: + logger.warning( + "No human input available for unresolved conflicts on incident %s; " + "logged for manual review", + report.incident_id, + ) + return None + + if response.strip().lower() in ("y", "yes"): + logger.info( + "Human accepted diagnosis for incident %s despite %d unresolved conflict(s)", + report.incident_id, + len(conflicts), + ) + return True + logger.warning( + "Human rejected diagnosis for incident %s due to unresolved conflicts", + report.incident_id, + ) + return False diff --git a/mcp-server/tests/test_escalation.py b/mcp-server/tests/test_escalation.py index 1f79e9a..545740a 100644 --- a/mcp-server/tests/test_escalation.py +++ b/mcp-server/tests/test_escalation.py @@ -1,9 +1,22 @@ """Tests for escalation logic.""" +from unittest.mock import MagicMock + import pytest +from streamops_mcp.agent import escalation from streamops_mcp.agent.escalation import escalate -from streamops_mcp.agent.schemas import IncidentReport, Severity +from streamops_mcp.agent.schemas import ( + ClaimRecord, + Confidence, + ConflictRecord, + DiagnosisReport, + IncidentReport, + RootCause, + Severity, + SourceRecord, +) +from streamops_mcp.agent.schemas.diagnosis import AffectedComponent from streamops_mcp.agent.schemas.incident import RecommendedAction @@ -29,8 +42,67 @@ def _make_report(severity: Severity) -> IncidentReport: ) -class TestEscalation: +def _make_diagnosis(*, conflict_resolution: str | None = "unresolved") -> DiagnosisReport: + """Build a valid diagnosis with two attributed claims. + + When conflict_resolution is given, adds a conflict between the two claims + with that resolution state; None means no conflict at all. + """ + conflicts = [] + if conflict_resolution is not None: + conflicts = [ + ConflictRecord( + conflict_id="conf-001", + topic="consumer lag value", + claim_a_id="C01", + claim_b_id="C02", + resolution=conflict_resolution, + notes="Prometheus and Kafka disagree on the lag", + ) + ] + return DiagnosisReport( + anomaly_type="latency_spike", + detected_at="2026-07-09T12:00:00Z", + sources=[ + SourceRecord( + source_id="src-001", + tool_name="query_flink_jobs", + retrieved_at="2026-07-09T12:00:00Z", + raw_output="{}", + ) + ], + claims=[ + ClaimRecord( + claim_id="C01", text="Lag is 45000", source_id="src-001", confidence=Confidence.HIGH + ), + ClaimRecord( + claim_id="C02", + text="Lag is 12000", + source_id="src-001", + confidence=Confidence.MEDIUM, + ), + ], + conflicts=conflicts, + affected_components=[ + AffectedComponent( + name="kafka-consumer", role="consumes", status="degraded", evidence="lag rising" + ) + ], + root_cause=RootCause(summary="Backpressure", confidence="medium", reasoning="chain"), + tools_used=["query_flink_jobs"], + ) + + +@pytest.fixture +def mock_audit(monkeypatch): + """Replace the module-level audit logger so tests inspect its call, not disk.""" + fake = MagicMock() + fake.log_incident = MagicMock(return_value={}) + monkeypatch.setattr(escalation, "_audit", fake) + return fake + +class TestEscalation: @pytest.mark.asyncio async def test_low_severity_logs_only(self, caplog): # Arrange @@ -105,3 +177,121 @@ async def test_critical_no_stdin(self, caplog, monkeypatch): # Assert assert "No human input" in caplog.text + + +class TestUnresolvedConflictEscalation: + """A diagnosis with unresolved conflicts must never pass silently.""" + + @pytest.mark.asyncio + async def test_unresolved_conflict_is_surfaced_and_gated(self, caplog, monkeypatch, mock_audit): + # Arrange: a MEDIUM incident whose diagnosis has an unresolved conflict + report = _make_report(Severity.MEDIUM) + diagnosis = _make_diagnosis(conflict_resolution="unresolved") + monkeypatch.setattr("builtins.input", lambda: "y") + + # Act + with caplog.at_level("INFO"): + await escalate(report, diagnosis=diagnosis) + + # Assert: surfaced at ERROR, human gate ran, and the audit records it + assert "UNRESOLVED" in caplog.text + assert "accepted diagnosis" in caplog.text + kwargs = mock_audit.log_incident.call_args.kwargs + assert kwargs["unresolved_conflict_count"] == 1 + assert kwargs["conflicts_acknowledged"] is True + + @pytest.mark.asyncio + async def test_unresolved_conflict_human_rejects(self, caplog, monkeypatch, mock_audit): + # Arrange + report = _make_report(Severity.MEDIUM) + diagnosis = _make_diagnosis(conflict_resolution="unresolved") + monkeypatch.setattr("builtins.input", lambda: "n") + + # Act + with caplog.at_level("INFO"): + await escalate(report, diagnosis=diagnosis) + + # Assert + assert "rejected diagnosis" in caplog.text + assert mock_audit.log_incident.call_args.kwargs["conflicts_acknowledged"] is False + + @pytest.mark.asyncio + async def test_unresolved_conflict_no_stdin_still_not_dropped( + self, caplog, monkeypatch, mock_audit + ): + # Arrange: automated run, no human available + report = _make_report(Severity.LOW) + diagnosis = _make_diagnosis(conflict_resolution="unresolved") + monkeypatch.setattr("builtins.input", lambda: (_ for _ in ()).throw(EOFError)) + + # Act + with caplog.at_level("WARNING"): + await escalate(report, diagnosis=diagnosis) + + # Assert: surfaced and audited even though nobody could acknowledge + assert "UNRESOLVED" in caplog.text + kwargs = mock_audit.log_incident.call_args.kwargs + assert kwargs["unresolved_conflict_count"] == 1 + assert kwargs["conflicts_acknowledged"] is None + + @pytest.mark.asyncio + async def test_resolved_conflict_is_not_gated(self, monkeypatch, mock_audit): + # Arrange: the conflict was resolved, so no gate and no prompt + report = _make_report(Severity.MEDIUM) + diagnosis = _make_diagnosis(conflict_resolution="resolved_a") + + def _fail_if_called(): + raise AssertionError("input() must not be called for a resolved conflict") + + monkeypatch.setattr("builtins.input", _fail_if_called) + + # Act + await escalate(report, diagnosis=diagnosis) + + # Assert + assert mock_audit.log_incident.call_args.kwargs["unresolved_conflict_count"] == 0 + + @pytest.mark.asyncio + async def test_no_conflicts_unchanged(self, monkeypatch, mock_audit): + # Arrange: diagnosis with no conflicts at all + report = _make_report(Severity.MEDIUM) + diagnosis = _make_diagnosis(conflict_resolution=None) + + def _fail_if_called(): + raise AssertionError("input() must not be called with no conflicts") + + monkeypatch.setattr("builtins.input", _fail_if_called) + + # Act + await escalate(report, diagnosis=diagnosis) + + # Assert + kwargs = mock_audit.log_incident.call_args.kwargs + assert kwargs["unresolved_conflict_count"] == 0 + assert kwargs["conflicts_acknowledged"] is None + + @pytest.mark.asyncio + async def test_critical_with_conflict_prompts_once_not_twice( + self, caplog, monkeypatch, mock_audit + ): + # Arrange: CRITICAL already gates; conflicts are surfaced but not double-prompted + report = _make_report(Severity.CRITICAL) + diagnosis = _make_diagnosis(conflict_resolution="unresolved") + calls = {"n": 0} + + def _count_input(): + calls["n"] += 1 + return "y" + + monkeypatch.setattr("builtins.input", _count_input) + + # Act + with caplog.at_level("ERROR"): + await escalate(report, diagnosis=diagnosis) + + # Assert: conflicts surfaced, single human prompt (the CRITICAL one) + assert "UNRESOLVED" in caplog.text + assert calls["n"] == 1 + kwargs = mock_audit.log_incident.call_args.kwargs + assert kwargs["unresolved_conflict_count"] == 1 + assert kwargs["conflicts_acknowledged"] is None # gated by severity, not separately