feat(agent): enforce unresolved-conflict escalation to the coordinator

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:

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

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