From 15a0565d3911964aa893053704874262a8765c0a Mon Sep 17 00:00:00 2001 From: matt butler Date: Thu, 9 Jul 2026 16:17:22 -0400 Subject: [PATCH] feat(agent): cross-cycle volatility gauge for incident dedup + change awareness Closes #77. The coordinator is stateless between cycles and never read its own history, so a persistent anomaly (latency stays bad for N cycles) was re-diagnosed and re-escalated every cycle (alert spam), and every sighting was treated as the first. This adds a lightweight in-memory cross-cycle memory on the coordinator instance (which lives for the whole run loop). New VolatilityGauge (agent/volatility.py): - fingerprints each anomaly as anomaly_type:component (idempotent dedup key), - classifies a detection as NEW / ONGOING / WORSENING against history, and marks other previously-active incidents RESOLVED, - WORSENING when observed_value rose by >= agent_incident_worsen_pct since last cycle; ONGOING recurrence within agent_incident_ongoing_gap cycles, else NEW, - decides should_report: NEW/WORSENING always; ONGOING only until reported once (agent_incident_dedup), then suppressed, - prior_context() feeds a "since last cycle" summary into the next detection so an ongoing anomaly is recognized as such (change awareness). run_cycle now stamps a cycle number, observes each detection, injects prior context into detection, suppresses a reported unchanged ongoing incident before the expensive diagnose/report, marks reported after escalation, and logs resolved incidents on all-clear. Thresholds externalized to config, validated at startup. Tests: 11 gauge unit tests (fingerprint, new/ongoing/worsening/gap, suppression, not-yet-reported still reports, resolved, prior_context, all-clear, dedup-off), a run_cycle two-cycle integration test proving a persistent incident reports once and is suppressed the second cycle, and config defaults/validation. Full suite 273 passed; ruff + mypy clean. --- mcp-server/src/streamops_mcp/agent/monitor.py | 45 ++++- .../src/streamops_mcp/agent/volatility.py | 178 ++++++++++++++++++ mcp-server/src/streamops_mcp/config.py | 12 ++ mcp-server/tests/test_config.py | 12 ++ mcp-server/tests/test_monitor.py | 42 +++++ mcp-server/tests/test_volatility.py | 153 +++++++++++++++ 6 files changed, 441 insertions(+), 1 deletion(-) create mode 100644 mcp-server/src/streamops_mcp/agent/volatility.py create mode 100644 mcp-server/tests/test_volatility.py diff --git a/mcp-server/src/streamops_mcp/agent/monitor.py b/mcp-server/src/streamops_mcp/agent/monitor.py index 41e72e5..9ec15e3 100644 --- a/mcp-server/src/streamops_mcp/agent/monitor.py +++ b/mcp-server/src/streamops_mcp/agent/monitor.py @@ -35,6 +35,7 @@ Severity, ) from streamops_mcp.agent.tools import ALL_TOOLS, DIAGNOSTIC_TOOLS +from streamops_mcp.agent.volatility import VolatilityGauge from streamops_mcp.config import config from streamops_mcp.logging_setup import ( new_correlation_id, @@ -116,6 +117,14 @@ def __init__(self, model: str | None = None, multi_agent: bool = True): self.model = model or config.agent_model self.multi_agent = multi_agent self.max_tool_rounds = config.agent_max_tool_rounds + # Cross-cycle memory (issue #77); persists for the life of this instance, + # which is the whole run loop. + self._cycle_count = 0 + self._gauge = VolatilityGauge( + ongoing_gap=config.agent_incident_ongoing_gap, + worsen_pct=config.agent_incident_worsen_pct, + dedup=config.agent_incident_dedup, + ) @staticmethod def _is_retryable(exc: Exception) -> bool: @@ -174,18 +183,45 @@ async def run_cycle(self) -> IncidentReport | None: """ cycle_id = new_correlation_id() token = set_correlation_id(cycle_id) + self._cycle_count += 1 + cycle = self._cycle_count try: logger.info( - "Starting monitoring cycle (cycle_id=%s, multi_agent=%s)", + "Starting monitoring cycle (cycle_id=%s, cycle=%d, multi_agent=%s)", cycle_id, + cycle, self.multi_agent, ) detection = await self._detect_anomalies() if detection is None: + resolved = self._gauge.note_all_clear(cycle) + if resolved: + logger.info("Resolved since last cycle: %s", ", ".join(resolved)) logger.info("No anomalies detected, infrastructure healthy") return None + # Cross-cycle memory (issue #77): classify this anomaly against history + # and suppress a persistent, already-reported incident so it is not + # re-escalated every cycle. + delta = self._gauge.observe(detection, cycle) + logger.info( + "Volatility: incident %s is %s (occurrence %d)", + delta.fingerprint, + delta.status.value, + delta.occurrences, + ) + if delta.resolved: + logger.info("Resolved since last cycle: %s", ", ".join(delta.resolved)) + if not delta.should_report: + logger.info( + "Suppressing duplicate escalation for ongoing unchanged incident %s " + "(active %d cycles)", + delta.fingerprint, + delta.occurrences, + ) + return None + if self.multi_agent: diagnosis = await self._run_diagnostics(detection) if diagnosis is None: @@ -238,6 +274,8 @@ async def run_cycle(self) -> IncidentReport | None: ) await escalate(report, diagnosis=diagnosis) + # Mark reported so an unchanged recurrence next cycle is suppressed (#77). + self._gauge.mark_reported(delta.fingerprint) return report finally: reset_correlation_id(token) @@ -541,10 +579,15 @@ async def _detect_anomalies(self) -> DetectedAnomaly | None: agent carries typed context instead of prose. """ detection_schema = DetectedAnomaly.model_json_schema() + # Change awareness (issue #77): tell detection what carried over from prior + # cycles so an ongoing anomaly is recognized as such, not a first sighting. + prior_context = self._gauge.prior_context(self._cycle_count) + continuity = f"{prior_context}\n\n" if prior_context else "" messages: list[dict[str, Any]] = [ { "role": "user", "content": ( + f"{continuity}" "Run a health check on the streaming infrastructure. Check Flink jobs, " "consumer lag, and recent events. If everything is healthy, say so briefly. " "If you detect an anomaly, respond with ONLY a JSON object matching this " diff --git a/mcp-server/src/streamops_mcp/agent/volatility.py b/mcp-server/src/streamops_mcp/agent/volatility.py new file mode 100644 index 0000000..7617d7a --- /dev/null +++ b/mcp-server/src/streamops_mcp/agent/volatility.py @@ -0,0 +1,178 @@ +"""Cross-cycle volatility gauge (issue #77). + +The coordinator is stateless between monitoring cycles: each cycle rebuilds +context and, without help, treats every sighting of a persistent anomaly as the +first. This gauge is a lightweight in-memory cross-cycle memory: it fingerprints +each detected anomaly, tracks it across cycles, classifies it as new / ongoing / +worsening / resolved, and decides whether it warrants re-escalation. + +Two jobs: + 1. Idempotency: a persistent, unchanged anomaly is diagnosed and reported once, + then suppressed until it materially changes or resolves (no alert spam). + 2. Change awareness: a human-readable "since last cycle" summary is fed into the + next detection so the agent knows an anomaly is ongoing/worsening, not new. + +State lives on the coordinator instance (which persists across the run loop), so +it resets on process restart. That is an accepted tradeoff for the gauge; durable +history already lives in the audit log. +""" + +import re +from dataclasses import dataclass, field +from enum import StrEnum + +from streamops_mcp.agent.schemas import DetectedAnomaly + +_LEADING_NUMBER = re.compile(r"[-+]?\d*\.?\d+") + + +class IncidentStatus(StrEnum): + """How a detected anomaly relates to what the gauge has seen before.""" + + NEW = "new" + ONGOING = "ongoing" + WORSENING = "worsening" + + +def _parse_value(value: str | None) -> float | None: + """Extract a leading number from an observed_value string ('2,340ms' -> 2340).""" + if value is None: + return None + match = _LEADING_NUMBER.search(value.replace(",", "")) + return float(match.group()) if match else None + + +@dataclass +class _IncidentMemory: + fingerprint: str + anomaly_type: str + component: str + first_cycle: int + last_cycle: int + occurrences: int + last_value: float | None + reported: bool = False + + +@dataclass +class VolatilityDelta: + """The cross-cycle verdict for one cycle's detected anomaly.""" + + status: IncidentStatus + fingerprint: str + occurrences: int + should_report: bool + resolved: list[str] = field(default_factory=list) + + +class VolatilityGauge: + """Tracks anomalies across cycles for dedup and change-awareness.""" + + def __init__( + self, + *, + ongoing_gap: int = 1, + worsen_pct: float = 0.25, + dedup: bool = True, + ): + self._ongoing_gap = ongoing_gap + self._worsen_pct = worsen_pct + self._dedup = dedup + self._memory: dict[str, _IncidentMemory] = {} + + @staticmethod + def fingerprint(anomaly: DetectedAnomaly) -> str: + """Stable dedup key for an anomaly: its type on its component.""" + component = anomaly.affected_component or "unknown" + return f"{anomaly.anomaly_type}:{component}" + + def _is_active(self, mem: _IncidentMemory, cycle: int) -> bool: + """Was this incident seen recently enough to still be the same occurrence?""" + return cycle - mem.last_cycle <= self._ongoing_gap + + def prior_context(self, cycle: int) -> str: + """A 'since last cycle' summary of still-active incidents, for detection. + + Empty string when there is nothing carried over (e.g. the first cycle). + """ + active = [ + m + for m in self._memory.values() + if cycle - m.last_cycle <= self._ongoing_gap and cycle > m.last_cycle + ] + if not active: + return "" + lines = [ + f"- {m.anomaly_type} on {m.component}: active {m.occurrences} cycle(s)" + + (f", last observed value {m.last_value:g}" if m.last_value is not None else "") + for m in sorted(active, key=lambda m: m.fingerprint) + ] + return ( + "Context from prior cycles (for continuity, confirm whether these persist):\n" + + "\n".join(lines) + ) + + def observe(self, anomaly: DetectedAnomaly, cycle: int) -> VolatilityDelta: + """Classify this cycle's anomaly against history, update memory, decide reporting. + + Also marks any other previously-active incident as resolved (a different + anomaly this cycle means the prior one is no longer the active symptom). + """ + fp = self.fingerprint(anomaly) + value = _parse_value(anomaly.observed_value) + + resolved = [ + m.fingerprint + for m in self._memory.values() + if m.fingerprint != fp and self._is_active(m, cycle - 1) + ] + + mem = self._memory.get(fp) + if mem is None or not self._is_active(mem, cycle): + # New incident, or the same fingerprint returning after a gap. + status = IncidentStatus.NEW + self._memory[fp] = _IncidentMemory( + fingerprint=fp, + anomaly_type=anomaly.anomaly_type, + component=anomaly.affected_component or "unknown", + first_cycle=cycle, + last_cycle=cycle, + occurrences=1, + last_value=value, + ) + mem = self._memory[fp] + else: + worsened = self._worsened(mem.last_value, value) + status = IncidentStatus.WORSENING if worsened else IncidentStatus.ONGOING + mem.occurrences += 1 + mem.last_cycle = cycle + mem.last_value = value + + should_report = ( + status in (IncidentStatus.NEW, IncidentStatus.WORSENING) + or not self._dedup + or not mem.reported + ) + return VolatilityDelta( + status=status, + fingerprint=fp, + occurrences=mem.occurrences, + should_report=should_report, + resolved=sorted(resolved), + ) + + def _worsened(self, prior: float | None, current: float | None) -> bool: + """True if the observed value rose by at least the worsening threshold.""" + if prior is None or current is None or prior <= 0: + return False + return (current - prior) / prior >= self._worsen_pct + + def mark_reported(self, fingerprint: str) -> None: + """Record that an incident was actually reported, so dedup can suppress repeats.""" + mem = self._memory.get(fingerprint) + if mem is not None: + mem.reported = True + + def note_all_clear(self, cycle: int) -> list[str]: + """A healthy cycle resolves every previously-active incident; return them.""" + return sorted(m.fingerprint for m in self._memory.values() if self._is_active(m, cycle - 1)) diff --git a/mcp-server/src/streamops_mcp/config.py b/mcp-server/src/streamops_mcp/config.py index 691439f..a2dd8bc 100644 --- a/mcp-server/src/streamops_mcp/config.py +++ b/mcp-server/src/streamops_mcp/config.py @@ -76,6 +76,18 @@ class StreamOpsConfig(BaseSettings): # Only takes effect when agent_diagnostic_forks > 1. agent_hypothesis_mode: Literal["static", "map", "llm"] = "map" + # Cross-cycle volatility gauge (issue #77): give the stateless coordinator + # memory across cycles so a persistent anomaly is diagnosed/reported once, not + # re-escalated every cycle (idempotency / anti-alert-spam). + # A re-detected anomaly counts as ONGOING (same incident) if last seen within + # this many cycles; a longer gap makes it a NEW incident again. + agent_incident_ongoing_gap: int = 1 + # An ongoing anomaly whose observed_value rose by at least this fraction since + # last cycle counts as WORSENING (re-reports despite dedup). + agent_incident_worsen_pct: float = 0.25 + # Suppress re-escalation of an unchanged, already-reported ongoing incident. + agent_incident_dedup: bool = True + # Input sanitization agent_sanitize_max_output_chars: int = 20_000 diff --git a/mcp-server/tests/test_config.py b/mcp-server/tests/test_config.py index 5d731b1..28264b0 100644 --- a/mcp-server/tests/test_config.py +++ b/mcp-server/tests/test_config.py @@ -17,6 +17,18 @@ def test_defaults(self): # Fork defaults preserve single-agent behavior; hypothesis mode defaults to map. assert cfg.agent_diagnostic_forks == 1 assert cfg.agent_hypothesis_mode == "map" + # Cross-cycle volatility gauge defaults (issue #77). + assert cfg.agent_incident_ongoing_gap == 1 + assert cfg.agent_incident_worsen_pct == 0.25 + assert cfg.agent_incident_dedup is True + + def test_invalid_worsen_pct_rejected(self, monkeypatch): + # Arrange: a non-numeric threshold must fail fast at startup + monkeypatch.setenv("STREAMOPS_AGENT_INCIDENT_WORSEN_PCT", "not-a-number") + + # Act + Assert + with pytest.raises(ValidationError): + StreamOpsConfig() def test_hypothesis_mode_override(self, monkeypatch): monkeypatch.setenv("STREAMOPS_AGENT_HYPOTHESIS_MODE", "llm") diff --git a/mcp-server/tests/test_monitor.py b/mcp-server/tests/test_monitor.py index d2921cd..21a20a6 100644 --- a/mcp-server/tests/test_monitor.py +++ b/mcp-server/tests/test_monitor.py @@ -29,6 +29,7 @@ SourceRecord, ) from streamops_mcp.agent.schemas.diagnosis import AffectedComponent +from streamops_mcp.agent.volatility import VolatilityGauge def _fake_response(status_code: int) -> httpx.Response: @@ -840,6 +841,9 @@ async def test_two_cycles_get_distinct_correlation_ids(self, agent): # Arrange from streamops_mcp.logging_setup import get_correlation_id + # This test is about correlation ids, not dedup: disable cross-cycle + # suppression so both identical cycles run all the way through (#77). + agent._gauge = VolatilityGauge(dedup=False) seen: list[str] = [] async def _capture(*args, **kwargs): @@ -1269,3 +1273,41 @@ async def test_run_cycle_does_not_delegate_empty_diagnosis(self, agent, monkeypa assert result is None spawn_report.assert_not_awaited() mock_escalate.assert_not_awaited() + + +class TestCrossCycleMemory: + """Cross-cycle dedup through run_cycle: a persistent incident reports once (#77).""" + + _DETECTION = json.dumps( + { + "anomaly_type": "latency_spike", + "summary": "processing latency 2,000ms", + "detected_at": "2026-07-09T12:00:00Z", + "affected_component": "processor", + } + ) + + @pytest.mark.asyncio + async def test_persistent_incident_suppressed_on_second_cycle(self, agent, monkeypatch): + # Arrange: single-agent fan-out; the same anomaly is detected both cycles + monkeypatch.setattr("streamops_mcp.agent.monitor.config.agent_diagnostic_forks", 1) + + with patch("streamops_mcp.agent.monitor.escalate", new_callable=AsyncMock) as mock_escalate: + # Act: cycle 1 detects, diagnoses, reports, escalates + agent.client = _mock_client( + _api_response(self._DETECTION), + _api_response(_DIAGNOSIS_JSON), + _api_response(_REPORT_JSON), + ) + first = await agent.run_cycle() + + # Act: cycle 2 detects the same incident; only detection is consumed + agent.client = _mock_client(_api_response(self._DETECTION)) + second = await agent.run_cycle() + + # Assert: reported once, suppressed the second cycle (no re-escalation) + assert isinstance(first, IncidentReport) + assert second is None + assert mock_escalate.await_count == 1 + # cycle 2 stopped after detection: no diagnostic/report calls + assert agent.client.messages.create.call_count == 1 diff --git a/mcp-server/tests/test_volatility.py b/mcp-server/tests/test_volatility.py new file mode 100644 index 0000000..8972f07 --- /dev/null +++ b/mcp-server/tests/test_volatility.py @@ -0,0 +1,153 @@ +"""Tests for the cross-cycle volatility gauge (issue #77).""" + +from streamops_mcp.agent.schemas import DetectedAnomaly +from streamops_mcp.agent.volatility import IncidentStatus, VolatilityGauge + + +def _anom(anomaly_type="latency_spike", component="processor", value=None) -> DetectedAnomaly: + return DetectedAnomaly( + anomaly_type=anomaly_type, + summary="an anomaly", + detected_at="2026-07-09T12:00:00Z", + affected_component=component, + observed_value=value, + ) + + +class TestVolatilityGauge: + def test_fingerprint_is_type_and_component(self): + # Arrange + gauge = VolatilityGauge() + + # Act + Assert + assert gauge.fingerprint(_anom("latency_spike", "processor")) == "latency_spike:processor" + assert gauge.fingerprint(_anom("latency_spike", None)) == "latency_spike:unknown" + + def test_first_sighting_is_new_and_reports(self): + # Arrange + gauge = VolatilityGauge() + + # Act + delta = gauge.observe(_anom(), 1) + + # Assert + assert delta.status == IncidentStatus.NEW + assert delta.occurrences == 1 + assert delta.should_report is True + + def test_reported_ongoing_is_suppressed(self): + # Arrange: reported on cycle 1 + gauge = VolatilityGauge() + first = gauge.observe(_anom(), 1) + gauge.mark_reported(first.fingerprint) + + # Act: same incident next cycle + second = gauge.observe(_anom(), 2) + + # Assert: recognized as ongoing and suppressed (idempotency) + assert second.status == IncidentStatus.ONGOING + assert second.occurrences == 2 + assert second.should_report is False + + def test_ongoing_not_yet_reported_still_reports(self): + # Arrange: cycle 1 was never marked reported (e.g. the pipeline aborted) + gauge = VolatilityGauge() + gauge.observe(_anom(), 1) + + # Act + second = gauge.observe(_anom(), 2) + + # Assert: still owed a report + assert second.status == IncidentStatus.ONGOING + assert second.should_report is True + + def test_worsening_reescalates_even_if_reported(self): + # Arrange: reported at value 2000 + gauge = VolatilityGauge(worsen_pct=0.25) + first = gauge.observe(_anom(value="2000"), 1) + gauge.mark_reported(first.fingerprint) + + # Act: value jumps 50% + second = gauge.observe(_anom(value="3000"), 2) + + # Assert + assert second.status == IncidentStatus.WORSENING + assert second.should_report is True + + def test_small_change_is_not_worsening_and_stays_suppressed(self): + # Arrange + gauge = VolatilityGauge(worsen_pct=0.25) + first = gauge.observe(_anom(value="2000"), 1) + gauge.mark_reported(first.fingerprint) + + # Act: value rises only 5% + second = gauge.observe(_anom(value="2100"), 2) + + # Assert + assert second.status == IncidentStatus.ONGOING + assert second.should_report is False + + def test_returns_as_new_after_gap(self): + # Arrange: seen on cycle 1, then absent + gauge = VolatilityGauge(ongoing_gap=1) + first = gauge.observe(_anom(), 1) + gauge.mark_reported(first.fingerprint) + + # Act: reappears on cycle 5 (gap of 4 > ongoing_gap) + later = gauge.observe(_anom(), 5) + + # Assert: treated as a fresh incident + assert later.status == IncidentStatus.NEW + assert later.occurrences == 1 + assert later.should_report is True + + def test_other_active_incident_marked_resolved(self): + # Arrange + gauge = VolatilityGauge() + first = gauge.observe(_anom("latency_spike", "processor"), 1) + gauge.mark_reported(first.fingerprint) + + # Act: a different anomaly this cycle + second = gauge.observe(_anom("backpressure", "sink"), 2) + + # Assert: the prior active incident is reported resolved + assert "latency_spike:processor" in second.resolved + + def test_prior_context_empty_then_summarizes_active(self): + # Arrange + gauge = VolatilityGauge() + + # Assert: nothing carried over on the first cycle + assert gauge.prior_context(1) == "" + + # Act + gauge.observe(_anom("latency_spike", "processor"), 1) + context = gauge.prior_context(2) + + # Assert + assert "latency_spike" in context + assert "processor" in context + + def test_note_all_clear_lists_active_incidents(self): + # Arrange + gauge = VolatilityGauge() + gauge.observe(_anom("latency_spike", "processor"), 1) + + # Act + resolved = gauge.note_all_clear(2) + + # Assert + assert resolved == ["latency_spike:processor"] + + def test_dedup_disabled_always_reports(self): + # Arrange + gauge = VolatilityGauge(dedup=False) + first = gauge.observe(_anom(), 1) + gauge.mark_reported(first.fingerprint) + + # Act + second = gauge.observe(_anom(), 2) + + # Assert: no suppression when dedup is off + assert second.status == IncidentStatus.ONGOING + assert second.should_report is True