feat(agent): cross-cycle volatility gauge for incident dedup + change awareness

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 "

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))

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
 

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

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