diff --git a/mcp-server/src/streamops_mcp/agent/monitor.py b/mcp-server/src/streamops_mcp/agent/monitor.py index 6dbbd4c..3c84908 100644 --- a/mcp-server/src/streamops_mcp/agent/monitor.py +++ b/mcp-server/src/streamops_mcp/agent/monitor.py @@ -207,6 +207,9 @@ def _fallback_report(diagnosis: DiagnosisReport) -> IncidentReport: timeline=[f"Detected at {diagnosis.detected_at}"], recommended_actions=[], monitoring_notes="Report agent was unavailable; review diagnosis data directly", + # Even on the fallback path the incident stays traceable (issue #88). + sources=diagnosis.sources, + supporting_claims=diagnosis.claims, ) @staticmethod @@ -535,13 +538,17 @@ def _parse_diagnosis(self, text: str) -> DiagnosisReport: ) def _parse_incident(self, text: str, diagnosis: DiagnosisReport) -> IncidentReport: - """Extract an IncidentReport from the report agent's response.""" + """Extract an IncidentReport from the report agent's response. + + The report is always re-attributed from the diagnosis so the incident + stays traceable to its evidence (see issue #88). + """ try: json_str = self._extract_json(text) - return IncidentReport.model_validate_json(json_str) + report = IncidentReport.model_validate_json(json_str) except Exception as e: logger.warning("Failed to parse IncidentReport: %s, using fallback", e) - return IncidentReport( + report = IncidentReport( incident_id=str(uuid.uuid4())[:8], title=f"Anomaly: {diagnosis.anomaly_type}", severity=Severity.MEDIUM, @@ -553,6 +560,21 @@ def _parse_incident(self, text: str, diagnosis: DiagnosisReport) -> IncidentRepo recommended_actions=[], monitoring_notes="Monitor after remediation", ) + return self._attach_attribution(report, diagnosis) + + @staticmethod + def _attach_attribution(report: IncidentReport, diagnosis: DiagnosisReport) -> IncidentReport: + """Carry the diagnosis's sources and claims into the report. + + Deterministic (not LLM-supplied), so no hallucinated source_ids; the + merged report is re-validated so the attribution invariant holds on what + the pipeline actually emits. The diagnosis is already validated, so this + never fails in the normal path. See issue #88. + """ + merged = report.model_dump() + merged["sources"] = [s.model_dump() for s in diagnosis.sources] + merged["supporting_claims"] = [c.model_dump() for c in diagnosis.claims] + return IncidentReport.model_validate(merged) def _extract_json(self, text: str) -> str: """Extract JSON from text that may contain markdown code blocks.""" diff --git a/mcp-server/src/streamops_mcp/agent/schemas/incident.py b/mcp-server/src/streamops_mcp/agent/schemas/incident.py index 5fe1e9d..b18fdc1 100644 --- a/mcp-server/src/streamops_mcp/agent/schemas/incident.py +++ b/mcp-server/src/streamops_mcp/agent/schemas/incident.py @@ -3,11 +3,17 @@ The Report Agent receives a DiagnosisReport and produces an IncidentReport with severity classification and recommended actions. This is what gets routed through the escalation logic. + +The report carries the diagnosis's sources and supporting claims forward so +the incident a human reads stays traceable to the tool signals behind it: +synthesis must not drop attribution. See issue #88. """ from enum import StrEnum -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator + +from streamops_mcp.agent.schemas.diagnosis import ClaimRecord, Confidence, SourceRecord class Severity(StrEnum): @@ -35,7 +41,9 @@ class IncidentReport(BaseModel): """ incident_id: str = Field(description="Unique identifier for this incident") - title: str = Field(description="Short, descriptive title (e.g., 'Checkpoint timeout on StreamOps Processor')") + title: str = Field( + description="Short, descriptive title (e.g., 'Checkpoint timeout on StreamOps Processor')" + ) severity: Severity = Field(description="Severity classification driving escalation routing") summary: str = Field(description="2-3 sentence executive summary for on-call") anomaly_type: str = Field(description="Category from the diagnosis") @@ -58,3 +66,41 @@ class IncidentReport(BaseModel): monitoring_notes: str = Field( description="What to watch after remediation to confirm resolution" ) + sources: list[SourceRecord] = Field( + default_factory=list, + description="Data sources carried from the diagnosis, so the incident is traceable", + ) + supporting_claims: list[ClaimRecord] = Field( + default_factory=list, + description="Attributed claims from the diagnosis that back this incident", + ) + + @model_validator(mode="after") + def _check_supporting_attribution(self): + """Keep the incident self-attributing: every supporting claim must trace + to a real source carried in this report. Mirrors the diagnosis-level + integrity check so attribution survives synthesis rather than dangling. + UNSOURCED claims are exempt, since they explicitly have no backing source. + """ + source_ids = [s.source_id for s in self.sources] + dup_sources = sorted({sid for sid in source_ids if source_ids.count(sid) > 1}) + if dup_sources: + raise ValueError( + f"Duplicate source_id(s) in incident report, references ambiguous: {dup_sources}" + ) + source_id_set = set(source_ids) + + claim_ids = [c.claim_id for c in self.supporting_claims] + dup_claims = sorted({cid for cid in claim_ids if claim_ids.count(cid) > 1}) + if dup_claims: + raise ValueError(f"Duplicate claim_id(s) in incident report: {dup_claims}") + + for claim in self.supporting_claims: + if claim.confidence == Confidence.UNSOURCED: + continue + if claim.source_id not in source_id_set: + raise ValueError( + f"Supporting claim '{claim.claim_id}' references unknown source_id " + f"'{claim.source_id}'; incident attribution would be untraceable" + ) + return self diff --git a/mcp-server/tests/test_monitor.py b/mcp-server/tests/test_monitor.py index 62831b1..7ac40b1 100644 --- a/mcp-server/tests/test_monitor.py +++ b/mcp-server/tests/test_monitor.py @@ -854,3 +854,33 @@ def _fresh_client(): # Assert: each cycle got its own id (no cross-cycle bleed) assert len(seen) == 2 assert seen[0] != seen[1] + + +class TestIncidentAttribution: + """The synthesized incident must stay traceable to the diagnosis (issue #88).""" + + def test_parse_incident_carries_diagnosis_attribution(self, agent): + # Arrange + diagnosis = DiagnosisReport.model_validate_json(_DIAGNOSIS_JSON) + + # Act: synthesize the report from the diagnosis + report = agent._parse_incident(_REPORT_JSON, diagnosis) + + # Assert: sources and claims survive synthesis, not just prose + assert [s.source_id for s in report.sources] == [s.source_id for s in diagnosis.sources] + assert [c.claim_id for c in report.supporting_claims] == [ + c.claim_id for c in diagnosis.claims + ] + # And the carried claim still traces to a real carried source + assert report.supporting_claims[0].source_id in {s.source_id for s in report.sources} + + def test_fallback_report_carries_attribution(self, agent): + # Arrange + diagnosis = DiagnosisReport.model_validate_json(_DIAGNOSIS_JSON) + + # Act: even when the report agent is unavailable + report = agent._fallback_report(diagnosis) + + # Assert + assert len(report.sources) == len(diagnosis.sources) + assert len(report.supporting_claims) == len(diagnosis.claims) diff --git a/mcp-server/tests/test_schemas.py b/mcp-server/tests/test_schemas.py index 4c0bca7..6adf0f1 100644 --- a/mcp-server/tests/test_schemas.py +++ b/mcp-server/tests/test_schemas.py @@ -733,6 +733,82 @@ def test_incident_defaults_to_empty_low_confidence(self): assert report.low_confidence_claims == [] +class TestIncidentReportAttribution: + """The incident must stay traceable: supporting claims trace to real sources.""" + + _BASE = { + "incident_id": "inc-001", + "title": "Latency spike", + "severity": "HIGH", + "summary": "Latency exceeded SLA.", + "anomaly_type": "latency_spike", + "root_cause": "GC pressure", + "affected_components": ["flink-operator"], + "timeline": ["15:00 - breach"], + "recommended_actions": [], + "monitoring_notes": "Watch heap", + } + + def _source(self, sid="src-001"): + return { + "source_id": sid, + "tool_name": "query_flink_jobs", + "retrieved_at": "2026-07-09T12:00:00Z", + "raw_output": "{}", + } + + def _claim(self, cid="C01", sid="src-001", confidence="HIGH"): + return {"claim_id": cid, "text": "Heap at 92%", "source_id": sid, "confidence": confidence} + + def test_valid_attribution_passes(self): + # Arrange + Act + report = IncidentReport.model_validate( + {**self._BASE, "sources": [self._source()], "supporting_claims": [self._claim()]} + ) + + # Assert + assert report.supporting_claims[0].source_id == "src-001" + + def test_defaults_to_empty_attribution(self): + # Arrange + Act: a report with no attribution is still valid (fallback path) + report = IncidentReport.model_validate(self._BASE) + + # Assert + assert report.sources == [] + assert report.supporting_claims == [] + + def test_supporting_claim_referencing_unknown_source_rejected(self): + # Arrange + Act + Assert + with pytest.raises(ValidationError, match="unknown source_id"): + IncidentReport.model_validate( + { + **self._BASE, + "sources": [self._source("src-001")], + "supporting_claims": [self._claim("C01", "src-999")], + } + ) + + def test_unsourced_supporting_claim_needs_no_source(self): + # Arrange + Act: UNSOURCED claims are exempt from the source requirement + report = IncidentReport.model_validate( + { + **self._BASE, + "sources": [], + "supporting_claims": [self._claim("C01", "none", "UNSOURCED")], + } + ) + + # Assert + assert report.supporting_claims[0].confidence.value == "UNSOURCED" + + def test_duplicate_source_id_rejected(self): + # Arrange + Act + Assert + with pytest.raises(ValidationError, match="Duplicate source_id"): + IncidentReport.model_validate( + {**self._BASE, "sources": [self._source("src-001"), self._source("src-001")]} + ) + + class TestDraftOnlyContract: def test_requires_human_approval_defaults_true(self): # Arrange