feat(agent): preserve claim-source attribution through report synthesis

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."""

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

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)

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