From 232e15365f892ab5eb559869b29334e41e5754f2 Mon Sep 17 00:00:00 2001 From: matt butler Date: Thu, 9 Jul 2026 13:38:14 -0400 Subject: [PATCH] feat(agent): enforce claim-source attribution integrity (#81) Add a model_validator to DiagnosisReport that enforces referential integrity on the claim-source-mapping pattern, so nothing in a diagnosis is untraceable: - every sourced claim's source_id must reference a real SourceRecord (UNSOURCED claims are exempt, they explicitly have no backing source) - every conflict must reference two real claims - source_ids and claim_ids must be unique, so references are unambiguous This closes the "attribution lost" failure the pattern exists to prevent: a dangling source_id would leave a finding with no traceable origin. 5 new tests. Closes #81 --- .../streamops_mcp/agent/schemas/diagnosis.py | 66 +++++++-- mcp-server/tests/test_schemas.py | 128 ++++++++++++++++-- 2 files changed, 173 insertions(+), 21 deletions(-) diff --git a/mcp-server/src/streamops_mcp/agent/schemas/diagnosis.py b/mcp-server/src/streamops_mcp/agent/schemas/diagnosis.py index 9ecbd0a..326fc45 100644 --- a/mcp-server/src/streamops_mcp/agent/schemas/diagnosis.py +++ b/mcp-server/src/streamops_mcp/agent/schemas/diagnosis.py @@ -7,7 +7,7 @@ from enum import StrEnum -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator class Confidence(StrEnum): @@ -29,7 +29,9 @@ class SourceRecord(BaseModel): """ source_id: str = Field(description="Unique identifier for this source (e.g., 'src-001')") - tool_name: str = Field(description="MCP tool that produced this data (e.g., 'query_flink_jobs')") + tool_name: str = Field( + description="MCP tool that produced this data (e.g., 'query_flink_jobs')" + ) retrieved_at: str = Field(description="ISO-8601 timestamp when the data was retrieved") raw_output: str = Field(description="Verbatim or summarized tool output") @@ -44,7 +46,9 @@ class ClaimRecord(BaseModel): """ claim_id: str = Field(description="Unique identifier for this claim (e.g., 'C01')") - text: str = Field(description="The factual claim (e.g., 'Consumer lag is 45,000 on partition 2')") + text: str = Field( + description="The factual claim (e.g., 'Consumer lag is 45,000 on partition 2')" + ) source_id: str = Field(description="References SourceRecord.source_id that produced this claim") confidence: Confidence = Field( description=( @@ -86,7 +90,9 @@ class AffectedComponent(BaseModel): name: str = Field(description="Component identifier (e.g., 'kafka-consumer', 'flink-operator')") role: str = Field(description="What this component does in the pipeline") status: str = Field(description="Current state: healthy, degraded, failing, unknown") - evidence: str = Field(description="Specific metric or log entry supporting the status assessment") + evidence: str = Field( + description="Specific metric or log entry supporting the status assessment" + ) class RootCause(BaseModel): @@ -111,7 +117,9 @@ class DiagnosisReport(BaseModel): """ - anomaly_type: str = Field(description="Category: latency_spike, throughput_drop, backpressure, checkpoint_failure, memory_pressure, error_burst") + anomaly_type: str = Field( + description="Category: latency_spike, throughput_drop, backpressure, checkpoint_failure, memory_pressure, error_burst" + ) detected_at: str = Field(description="ISO-8601 timestamp when the anomaly was first detected") sources: list[SourceRecord] = Field( default_factory=list, @@ -129,10 +137,52 @@ class DiagnosisReport(BaseModel): description="Components involved in or affected by the anomaly" ) root_cause: RootCause = Field(description="Identified or suspected root cause") - tools_used: list[str] = Field( - description="MCP tools called during investigation (audit trail)" - ) + tools_used: list[str] = Field(description="MCP tools called during investigation (audit trail)") raw_evidence: list[str] = Field( default_factory=list, description="Key data points collected during investigation", ) + + @model_validator(mode="after") + def _check_attribution_integrity(self): + """Enforce claim-source attribution integrity so nothing is untraceable. + + Every sourced claim must reference a real source, and every conflict must + reference real claims. This prevents the "attribution lost" failure the + claim-source pattern exists to stop: a dangling source_id would leave a + finding with no traceable origin. UNSOURCED claims are exempt, since they + explicitly have no backing data 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), references would be ambiguous: {dup_sources}" + ) + source_id_set = set(source_ids) + + claim_ids = [c.claim_id for c in self.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), conflict references would be ambiguous: {dup_claims}" + ) + claim_id_set = set(claim_ids) + + for claim in self.claims: + if claim.confidence == Confidence.UNSOURCED: + continue + if claim.source_id not in source_id_set: + raise ValueError( + f"Claim '{claim.claim_id}' references unknown source_id " + f"'{claim.source_id}'; attribution would be untraceable" + ) + + for conflict in self.conflicts: + for cid in (conflict.claim_a_id, conflict.claim_b_id): + if cid not in claim_id_set: + raise ValueError( + f"Conflict '{conflict.conflict_id}' references unknown claim_id '{cid}'" + ) + + return self diff --git a/mcp-server/tests/test_schemas.py b/mcp-server/tests/test_schemas.py index 8e4ed00..cf9f94f 100644 --- a/mcp-server/tests/test_schemas.py +++ b/mcp-server/tests/test_schemas.py @@ -3,6 +3,7 @@ import json import pytest +from pydantic import ValidationError from streamops_mcp.agent.schemas import ( ClaimRecord, @@ -20,7 +21,6 @@ class TestDiagnosisReport: - def test_valid_diagnosis(self): # Arrange data = { @@ -113,7 +113,6 @@ def test_round_trip_serialization(self): class TestIncidentReport: - def test_valid_incident(self): # Arrange data = { @@ -163,7 +162,6 @@ def test_json_schema_includes_severity_enum(self): class TestRecommendedAction: - def test_action_with_downtime(self): # Arrange action = RecommendedAction( @@ -191,7 +189,6 @@ def test_action_no_downtime(self): class TestSourceRecord: - def test_valid_source(self): # Arrange + Act source = SourceRecord( @@ -207,7 +204,6 @@ def test_valid_source(self): class TestClaimRecord: - def test_valid_claim(self): # Arrange + Act claim = ClaimRecord( @@ -242,8 +238,116 @@ def test_claim_references_source(self): assert claim.confidence == Confidence.MEDIUM -class TestConflictRecord: +class TestAttributionIntegrity: + """DiagnosisReport enforces claim-source and conflict referential integrity (#81).""" + + @staticmethod + def _report(sources, claims, conflicts): + return dict( + anomaly_type="latency_spike", + detected_at="2026-06-18T15:00:00Z", + affected_components=[], + root_cause=RootCause(summary="s", confidence="low", reasoning="r"), + tools_used=["query_metrics"], + sources=sources, + claims=claims, + conflicts=conflicts, + ) + + def test_valid_attribution_passes(self): + # Arrange + sources = [ + SourceRecord( + source_id="src-001", + tool_name="query_metrics", + retrieved_at="2026-06-18T15:00:00Z", + raw_output="lag=45000", + ) + ] + claims = [ + ClaimRecord( + claim_id="C01", + text="Consumer lag is 45000", + source_id="src-001", + confidence=Confidence.HIGH, + ), + ClaimRecord( + claim_id="C02", + text="Backpressure rising", + source_id="src-001", + confidence=Confidence.MEDIUM, + ), + ] + conflicts = [ + ConflictRecord( + conflict_id="conf-001", topic="lag trend", claim_a_id="C01", claim_b_id="C02" + ) + ] + + # Act + report = DiagnosisReport(**self._report(sources, claims, conflicts)) + + # Assert + assert len(report.claims) == 2 + def test_claim_referencing_unknown_source_rejected(self): + # Arrange + sources = [ + SourceRecord( + source_id="src-001", tool_name="query_metrics", retrieved_at="t", raw_output="o" + ) + ] + claims = [ + ClaimRecord(claim_id="C01", text="x", source_id="src-999", confidence=Confidence.HIGH) + ] + + # Act / Assert + with pytest.raises(ValidationError, match="unknown source_id"): + DiagnosisReport(**self._report(sources, claims, [])) + + def test_unsourced_claim_needs_no_source(self): + # Arrange: an UNSOURCED claim legitimately has no backing source + claims = [ + ClaimRecord( + claim_id="C01", text="speculative", source_id="", confidence=Confidence.UNSOURCED + ) + ] + + # Act + report = DiagnosisReport(**self._report([], claims, [])) + + # Assert + assert report.claims[0].confidence == Confidence.UNSOURCED + + def test_conflict_referencing_unknown_claim_rejected(self): + # Arrange + sources = [ + SourceRecord(source_id="src-001", tool_name="t", retrieved_at="t", raw_output="o") + ] + claims = [ + ClaimRecord(claim_id="C01", text="x", source_id="src-001", confidence=Confidence.HIGH) + ] + conflicts = [ + ConflictRecord(conflict_id="conf-001", topic="t", claim_a_id="C01", claim_b_id="C99") + ] + + # Act / Assert + with pytest.raises(ValidationError, match="unknown claim_id"): + DiagnosisReport(**self._report(sources, claims, conflicts)) + + def test_duplicate_source_id_rejected(self): + # Arrange + sources = [ + SourceRecord(source_id="src-001", tool_name="t", retrieved_at="t", raw_output="a"), + SourceRecord(source_id="src-001", tool_name="t", retrieved_at="t", raw_output="b"), + ] + + # Act / Assert + with pytest.raises(ValidationError, match="Duplicate source_id"): + DiagnosisReport(**self._report(sources, [], [])) + + +class TestConflictRecord: def test_valid_conflict(self): # Arrange + Act conflict = ConflictRecord( @@ -286,7 +390,6 @@ def test_conflict_with_notes(self): class TestDiagnosisWithAttribution: - def test_full_diagnosis_with_claims_and_sources(self): # Arrange data = { @@ -449,7 +552,6 @@ def test_round_trip_with_attribution(self): class TestConfidence: - def test_confidence_enum_values(self): # Assert assert Confidence.HIGH.value == "HIGH" @@ -565,7 +667,9 @@ def test_mixed_confidence_claims_in_diagnosis(self): # Act high_claims = [c for c in report.claims if c.confidence == Confidence.HIGH] - low_claims = [c for c in report.claims if c.confidence in (Confidence.LOW, Confidence.UNSOURCED)] + low_claims = [ + c for c in report.claims if c.confidence in (Confidence.LOW, Confidence.UNSOURCED) + ] # Assert assert len(high_claims) == 1 @@ -573,7 +677,6 @@ def test_mixed_confidence_claims_in_diagnosis(self): class TestIncidentReportLowConfidence: - def test_incident_with_low_confidence_claims(self): # Arrange data = { @@ -630,7 +733,6 @@ def test_incident_defaults_to_empty_low_confidence(self): class TestDraftOnlyContract: - def test_requires_human_approval_defaults_true(self): # Arrange data = { @@ -683,7 +785,6 @@ def test_requires_human_approval_in_json_schema(self): class TestMonitorToDiagnosticHandoff: - def test_valid_handoff(self): # Arrange + Act handoff = MonitorToDiagnosticHandoff( @@ -699,6 +800,7 @@ def test_truncates_oversized_context(self, monkeypatch): # Arrange monkeypatch.setenv("STREAMOPS_AGENT_HANDOFF_MAX_CONTEXT_CHARS", "100") from streamops_mcp.config import StreamOpsConfig + test_config = StreamOpsConfig() monkeypatch.setattr("streamops_mcp.agent.schemas.handoff.config", test_config) @@ -729,7 +831,6 @@ def test_round_trip_serialization(self): class TestDiagnosticToReportHandoff: - def test_valid_handoff(self): # Arrange diagnosis = DiagnosisReport( @@ -761,6 +862,7 @@ def test_rejects_oversized_diagnosis(self, monkeypatch): # Arrange monkeypatch.setenv("STREAMOPS_AGENT_HANDOFF_MAX_CONTEXT_CHARS", "50") from streamops_mcp.config import StreamOpsConfig + test_config = StreamOpsConfig() monkeypatch.setattr("streamops_mcp.agent.schemas.handoff.config", test_config)