From db836389917d8c2f9d9b779a4481b72cd54db5db Mon Sep 17 00:00:00 2001 From: matt butler Date: Thu, 9 Jul 2026 14:55:43 -0400 Subject: [PATCH] feat(agent): preserve claim-source attribution through report synthesis Closes #88. Attribution was enforced within the diagnosis (#81) and passed intact into the Report agent, but lost at synthesis: IncidentReport carried no sources and no claims, so the artifact a human reads (prose root_cause/summary, text-only low_confidence_claims) could not be traced back to the tool signals behind it. Traceability survived only in the audit log's separate diagnosis dump. This violated the assertion that every synthesis step must output claim-source pairs. - IncidentReport gains sources: list[SourceRecord] and supporting_claims: list[ClaimRecord] (default empty), plus a self-contained validator mirroring the diagnosis integrity check: every non-UNSOURCED supporting claim traces to a real carried source; source_ids and claim_ids unique. - _parse_incident re-attributes every report from the diagnosis deterministically (_attach_attribution: carry sources + claims, re-validate), so no hallucinated source_ids and the invariant holds on what the pipeline emits. - _fallback_report carries attribution too, so the fallback path stays traceable. Root-cause/action-level provenance (citing specific claim_ids) is left as a follow-up; this delivers the deterministic no-loss guarantee first. Tests: IncidentReport attribution validator (valid, unknown-source rejected, UNSOURCED exempt, duplicate rejected, empty default) and monitor carry-forward (_parse_incident + _fallback_report). Full suite 236 passed; ruff + mypy clean. --- mcp-server/src/streamops_mcp/agent/monitor.py | 28 ++++++- .../streamops_mcp/agent/schemas/incident.py | 50 +++++++++++- mcp-server/tests/test_monitor.py | 30 ++++++++ mcp-server/tests/test_schemas.py | 76 +++++++++++++++++++ 4 files changed, 179 insertions(+), 5 deletions(-) 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