diff --git a/docs/fork-diagnostics-runbook.md b/docs/fork-diagnostics-runbook.md index 731bbef..bac40eb 100644 --- a/docs/fork-diagnostics-runbook.md +++ b/docs/fork-diagnostics-runbook.md @@ -67,18 +67,23 @@ correlated-logs feature, issue #84). The log format is: ``` In fork mode you will see the fan-out, one spawn line per fork with its -hypothesis, and the merge: +hypothesis, and the merge. By default (`agent_hypothesis_mode = map`) the +hypotheses are tailored to the anomaly type; a `latency_spike` fans out into its +latency-specific candidate causes: ``` [streamops-mcp.monitor] [cid=cyc-a1b2c3d4e5f6] INFO Fanning out 3 diagnostic forks -[streamops-mcp.monitor] [cid=cyc-a1b2c3d4e5f6] INFO Spawning Diagnostic Agent (hypothesis: Resource saturation: CPU, memory/heap, or GC pressure on the affected component.) -[streamops-mcp.monitor] [cid=cyc-a1b2c3d4e5f6] INFO Spawning Diagnostic Agent (hypothesis: Data-side cause: partition skew, hot keys, or a surge in input volume.) -[streamops-mcp.monitor] [cid=cyc-a1b2c3d4e5f6] INFO Spawning Diagnostic Agent (hypothesis: External dependency: a downstream sink, source, or coordination service degrading.) +[streamops-mcp.monitor] [cid=cyc-a1b2c3d4e5f6] INFO Spawning Diagnostic Agent (hypothesis: GC/heap pressure or long stop-the-world pauses on the TaskManager.) +[streamops-mcp.monitor] [cid=cyc-a1b2c3d4e5f6] INFO Spawning Diagnostic Agent (hypothesis: Serialization/deserialization cost or an expensive operator on the hot path.) +[streamops-mcp.monitor] [cid=cyc-a1b2c3d4e5f6] INFO Spawning Diagnostic Agent (hypothesis: Latency in an external call (enrichment, sink, or lookup) blocking the pipeline.) [streamops-mcp.monitor] [cid=cyc-a1b2c3d4e5f6] INFO Monitor->Diagnostic handoff validated (type=latency_spike, 412 chars) ... [streamops-mcp.monitor] [cid=cyc-a1b2c3d4e5f6] INFO Claim confidence distribution: 4 HIGH, 2 MEDIUM, 1 LOW, 0 UNSOURCED ``` +An ambiguous anomaly (`type=unknown`) instead fans out into the generic +investigative angles. See "Choosing hypotheses" below. + If the forks disagree on the root cause, the coordinator logs the cross-fork conflict (note the `xf-` conflict id and the `f{i}:` namespaced claim ids), and escalation surfaces it: @@ -148,6 +153,27 @@ conflict count: with `xf--` ids and the topic `cross-fork root-cause disagreement`, distinct from intra-fork conflicts. +## Choosing hypotheses + +How the fork hypotheses are chosen is set by `agent_hypothesis_mode` (issue #91), +and only matters when `agent_diagnostic_forks > 1`: + +- **`map`** (default): hypotheses tailored to the `anomaly_type` (latency_spike, + throughput_drop, backpressure, checkpoint_failure, memory_pressure, + error_burst), falling back to generic angles for an unknown type. No extra + LLM call. +- **`static`**: the fixed generic investigative angles regardless of type. +- **`llm`**: a cheap pre-fan-out call generates candidate hypotheses for the + specific anomaly; on any failure it falls back to `map`. + +The fork count is **adaptive**: it follows the number of hypotheses the anomaly +warrants (bounded by `agent_diagnostic_forks`), so a clear-cut anomaly with a +single plausible cause runs one agent even when the cap is higher. + +``` +export STREAMOPS_AGENT_HYPOTHESIS_MODE=llm # or map (default), or static +``` + ## Baseline to diff against Run the same scenario with `STREAMOPS_AGENT_DIAGNOSTIC_FORKS=1`. You should see: @@ -162,8 +188,7 @@ attributed diagnosis vs a single line of reasoning) is the feature. ## Notes -- The hypotheses are currently a fixed set of generic investigative angles. - Deriving them from the specific anomaly and adapting the fork count to - ambiguity is tracked in issue #91. +- Hypotheses are derived from the anomaly and the fork count adapts to + ambiguity (issue #91); see "Choosing hypotheses" above. - Cross-cycle change awareness ("what changed since last cycle") is tracked separately in issue #77. diff --git a/mcp-server/src/streamops_mcp/agent/monitor.py b/mcp-server/src/streamops_mcp/agent/monitor.py index ea67fa6..00b7383 100644 --- a/mcp-server/src/streamops_mcp/agent/monitor.py +++ b/mcp-server/src/streamops_mcp/agent/monitor.py @@ -49,9 +49,9 @@ DIAGNOSTIC_SYSTEM_PROMPT = load_prompt("diagnostic") REPORT_SYSTEM_PROMPT = load_prompt("report") -# Distinct investigative angles seeded into parallel diagnostic forks (issue #67) -# so each explores a different candidate cause instead of one line of reasoning -# tunnel-visioning on an ambiguous anomaly. A fan-out of N takes the first N. +# Generic investigative angles, used for ambiguous (unknown-type) anomalies and +# for "static" hypothesis mode. Each fork explores a different candidate cause +# instead of one line of reasoning tunnel-visioning (issue #67). DIAGNOSTIC_HYPOTHESES = [ "Resource saturation: CPU, memory/heap, or GC pressure on the affected component.", "Data-side cause: partition skew, hot keys, or a surge in input volume.", @@ -59,6 +59,42 @@ "Configuration or deployment change: a recent config, scaling, or version change.", ] +# Hypotheses tailored per anomaly_type (issue #91): forks investigate the causes +# actually plausible for this symptom rather than generic angles. The number of +# entries drives the adaptive fork count for a typed anomaly (bounded by the cap). +ANOMALY_HYPOTHESES = { + "latency_spike": [ + "GC/heap pressure or long stop-the-world pauses on the TaskManager.", + "Serialization/deserialization cost or an expensive operator on the hot path.", + "Latency in an external call (enrichment, sink, or lookup) blocking the pipeline.", + ], + "throughput_drop": [ + "A slow or stuck consumer/operator reducing end-to-end throughput.", + "A partition rebalance or reassignment interrupting consumption.", + "An upstream volume surge or source slowdown changing input rate.", + ], + "backpressure": [ + "Downstream sink saturation (slow writes) propagating backpressure upstream.", + "Operator skew or a hot key overloading one subtask.", + "Insufficient parallelism for the current load.", + ], + "checkpoint_failure": [ + "State size growth making checkpoints exceed their timeout.", + "Checkpoint timeout/interval configuration too tight for the workload.", + "State-backend or durable-storage I/O errors or slowness.", + ], + "memory_pressure": [ + "Heap exhaustion or GC thrash from object churn.", + "Unbounded state growth (missing TTL or retention).", + "Data skew concentrating memory on one subtask.", + ], + "error_burst": [ + "Bad or poison input (malformed events, schema drift) triggering exceptions.", + "A downstream dependency failing and surfacing as errors.", + "A recent deploy or config change regressing behavior.", + ], +} + # Confidence ordering for picking a fork's representative claim and the primary fork. _CONFIDENCE_RANK = { Confidence.HIGH: 3, @@ -205,27 +241,30 @@ async def run_cycle(self) -> IncidentReport | None: reset_correlation_id(token) async def _run_diagnostics(self, anomaly: DetectedAnomaly) -> DiagnosisReport | None: - """Diagnose the anomaly, single-agent or fork-style (issue #67). - - With ``agent_diagnostic_forks <= 1`` this is the original single - Diagnostic agent. With more, it fans out N agents concurrently, each - seeded with a distinct hypothesis, then merges the survivors. Returns - None only if the diagnosis could not be produced at all (single agent - failed, or every fork failed after retries). + """Diagnose the anomaly, single-agent or fork-style (issues #67, #91). + + The hypotheses are planned from the anomaly (see ``_plan_hypotheses``), + which also sets the fork count adaptively: a clear-cut anomaly runs one + agent, an ambiguous one fans out, never above ``agent_diagnostic_forks``. + With 0 or 1 planned hypotheses this is the original single Diagnostic + agent (seeded with the one hypothesis if there is one). Returns None only + if the diagnosis could not be produced at all (the single agent failed, + or every fork failed after retries). """ - fork_count = min(config.agent_diagnostic_forks, len(DIAGNOSTIC_HYPOTHESES)) + hypotheses = await self._plan_hypotheses(anomaly) - if fork_count <= 1: + if len(hypotheses) <= 1: + seed = hypotheses[0] if hypotheses else None try: return await self._retry_subagent( "Diagnostic Agent", - lambda: self._spawn_diagnostic_agent(anomaly), + lambda: self._spawn_diagnostic_agent(anomaly, hypothesis=seed), ) except Exception as exc: logger.error("Diagnostic Agent failed after retries, cycle aborted: %s", exc) return None - hypotheses = DIAGNOSTIC_HYPOTHESES[:fork_count] + fork_count = len(hypotheses) logger.info("Fanning out %d diagnostic forks", fork_count) async def _fork(index: int, hypothesis: str) -> DiagnosisReport: @@ -254,6 +293,67 @@ async def _fork(index: int, hypothesis: str) -> DiagnosisReport: ) return self._merge_fork_diagnoses(survivors) + async def _plan_hypotheses(self, anomaly: DetectedAnomaly) -> list[str]: + """Choose the diagnostic hypotheses (and thus the fork count) for an anomaly. + + Bounded by ``agent_diagnostic_forks``: <= 1 means single-agent (empty + list). Otherwise the pool depends on ``agent_hypothesis_mode``: + - "static": generic investigative angles. + - "map": angles tailored to the anomaly_type, generic for unknown types + (issue #91). The size of the type's entry adapts the fork count, so a + clear-cut type with one plausible cause runs one agent. + - "llm": angles generated per-anomaly, falling back to the map on failure. + """ + cap = config.agent_diagnostic_forks + if cap <= 1: + return [] + + mode = config.agent_hypothesis_mode + if mode == "llm": + generated = await self._generate_hypotheses_llm(anomaly, cap) + pool = generated or self._mapped_hypotheses(anomaly) + elif mode == "static": + pool = DIAGNOSTIC_HYPOTHESES + else: # "map" + pool = self._mapped_hypotheses(anomaly) + + return pool[:cap] + + @staticmethod + def _mapped_hypotheses(anomaly: DetectedAnomaly) -> list[str]: + """Per-type hypotheses, or the generic angles for an unknown/unmapped type.""" + return ANOMALY_HYPOTHESES.get(anomaly.anomaly_type) or DIAGNOSTIC_HYPOTHESES + + async def _generate_hypotheses_llm(self, anomaly: DetectedAnomaly, k: int) -> list[str] | None: + """Ask the model for up to k candidate root-cause hypotheses for this anomaly. + + Returns None on any failure so the caller falls back to the static map; + a diagnosis must never hinge on this optional pre-step succeeding. + """ + prompt = ( + f"Given this streaming-pipeline anomaly, list up to {k} distinct candidate " + "root-cause hypotheses worth investigating in parallel, most likely first, " + "one per line, no numbering or commentary.\n\n" + f"{anomaly.model_dump_json(indent=2)}" + ) + try: + response = await self.client.messages.create( + model=self.model, + max_tokens=config.agent_max_tokens, + messages=[{"role": "user", "content": prompt}], + ) + except Exception as exc: + logger.warning("Hypothesis generation failed (%s); falling back to the map", exc) + return None + + text = "".join(b.text for b in response.content if b.type == "text") + lines = [line.strip(" -*\t").strip() for line in text.splitlines()] + hypotheses = [line for line in lines if line][:k] + if not hypotheses: + logger.warning("Hypothesis generation returned nothing usable; falling back to the map") + return None + return hypotheses + @classmethod def _merge_fork_diagnoses(cls, diagnoses: list[DiagnosisReport]) -> DiagnosisReport: """Merge parallel fork diagnoses into one, preserving attribution. diff --git a/mcp-server/src/streamops_mcp/config.py b/mcp-server/src/streamops_mcp/config.py index 37d7b77..691439f 100644 --- a/mcp-server/src/streamops_mcp/config.py +++ b/mcp-server/src/streamops_mcp/config.py @@ -1,3 +1,5 @@ +from typing import Literal + from pydantic_settings import BaseSettings @@ -60,12 +62,20 @@ class StreamOpsConfig(BaseSettings): agent_max_retries: int = 2 agent_retry_base_delay: float = 1.0 - # Fork-style parallel diagnostic exploration (issue #67). Number of diagnostic - # sub-agents to fan out concurrently, each seeded with a distinct hypothesis. - # Default 1 preserves single-agent behavior; >1 fans out (capped at the number - # of defined hypotheses). Fan-out adds cost/latency, so raise it deliberately. + # Fork-style parallel diagnostic exploration (issue #67). Upper bound on the + # number of diagnostic sub-agents fanned out concurrently, each seeded with a + # distinct hypothesis. Default 1 preserves single-agent behavior; >1 enables + # fan-out. The actual count adapts to how many hypotheses the anomaly warrants + # (issue #91), never exceeding this cap. Fan-out adds cost/latency. agent_diagnostic_forks: int = 1 + # How diagnostic-fork hypotheses are chosen (issue #91): + # "static" - a fixed set of generic investigative angles + # "map" - hypotheses tailored to the anomaly_type (default; no extra LLM call) + # "llm" - generated per-anomaly by a cheap pre-fan-out LLM call + # Only takes effect when agent_diagnostic_forks > 1. + agent_hypothesis_mode: Literal["static", "map", "llm"] = "map" + # 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 05c20d4..5d731b1 100644 --- a/mcp-server/tests/test_config.py +++ b/mcp-server/tests/test_config.py @@ -1,11 +1,12 @@ """Tests for StreamOps config.""" +import pytest +from pydantic import ValidationError from streamops_mcp.config import StreamOpsConfig class TestConfig: - def test_defaults(self): cfg = StreamOpsConfig() assert cfg.flink_url == "http://localhost:8081" @@ -13,6 +14,22 @@ def test_defaults(self): assert cfg.kafka_bootstrap == "localhost:9092" assert cfg.kafka_events_topic == "stream-events" assert cfg.kafka_alerts_topic == "stream-alerts" + # Fork defaults preserve single-agent behavior; hypothesis mode defaults to map. + assert cfg.agent_diagnostic_forks == 1 + assert cfg.agent_hypothesis_mode == "map" + + def test_hypothesis_mode_override(self, monkeypatch): + monkeypatch.setenv("STREAMOPS_AGENT_HYPOTHESIS_MODE", "llm") + cfg = StreamOpsConfig() + assert cfg.agent_hypothesis_mode == "llm" + + def test_invalid_hypothesis_mode_rejected(self, monkeypatch): + # Arrange: an unknown mode must fail fast at startup, not silently pass + monkeypatch.setenv("STREAMOPS_AGENT_HYPOTHESIS_MODE", "bogus") + + # Act + Assert + with pytest.raises(ValidationError): + StreamOpsConfig() def test_env_override(self, monkeypatch): monkeypatch.setenv("STREAMOPS_FLINK_URL", "http://flink-prod:8081") diff --git a/mcp-server/tests/test_monitor.py b/mcp-server/tests/test_monitor.py index c26a5dd..137793e 100644 --- a/mcp-server/tests/test_monitor.py +++ b/mcp-server/tests/test_monitor.py @@ -13,7 +13,11 @@ import httpx import pytest -from streamops_mcp.agent.monitor import MonitorAgent +from streamops_mcp.agent.monitor import ( + ANOMALY_HYPOTHESES, + DIAGNOSTIC_HYPOTHESES, + MonitorAgent, +) from streamops_mcp.agent.schemas import ( ClaimRecord, Confidence, @@ -1064,3 +1068,130 @@ async def test_run_cycle_fork_path_end_to_end(self, agent, monkeypatch): assert isinstance(result, IncidentReport) assert agent.client.messages.create.call_count == 5 mock_escalate.assert_awaited_once() + + +class TestDynamicHypotheses: + """Hypotheses are derived from the anomaly, and the fork count adapts (issue #91).""" + + def _set(self, monkeypatch, *, forks, mode="map"): + monkeypatch.setattr("streamops_mcp.agent.monitor.config.agent_diagnostic_forks", forks) + monkeypatch.setattr("streamops_mcp.agent.monitor.config.agent_hypothesis_mode", mode) + + @pytest.mark.asyncio + async def test_cap_one_returns_no_hypotheses(self, agent, monkeypatch): + # Arrange: default cap keeps single-agent behavior + self._set(monkeypatch, forks=1) + + # Act + result = await agent._plan_hypotheses(_anom("latency_spike")) + + # Assert + assert result == [] + + @pytest.mark.asyncio + async def test_map_mode_uses_type_specific_hypotheses(self, agent, monkeypatch): + # Arrange + self._set(monkeypatch, forks=3, mode="map") + + # Act + result = await agent._plan_hypotheses(_anom("checkpoint_failure")) + + # Assert: tailored to the anomaly type, not the generic angles + assert result == ANOMALY_HYPOTHESES["checkpoint_failure"][:3] + assert any("state size" in h.lower() for h in result) + + @pytest.mark.asyncio + async def test_map_mode_unknown_falls_back_to_generic(self, agent, monkeypatch): + # Arrange: an ambiguous anomaly has no tailored map entry + self._set(monkeypatch, forks=3, mode="map") + + # Act + result = await agent._plan_hypotheses(_anom("unknown")) + + # Assert + assert result == DIAGNOSTIC_HYPOTHESES[:3] + + @pytest.mark.asyncio + async def test_static_mode_uses_generic(self, agent, monkeypatch): + # Arrange: static mode ignores the type map + self._set(monkeypatch, forks=2, mode="static") + + # Act + result = await agent._plan_hypotheses(_anom("latency_spike")) + + # Assert + assert result == DIAGNOSTIC_HYPOTHESES[:2] + + @pytest.mark.asyncio + async def test_cap_bounds_hypothesis_count(self, agent, monkeypatch): + # Arrange: cap below the number of mapped hypotheses + self._set(monkeypatch, forks=2, mode="map") + + # Act + result = await agent._plan_hypotheses(_anom("latency_spike")) + + # Assert + assert len(result) == 2 + + @pytest.mark.asyncio + async def test_llm_mode_uses_generated_hypotheses(self, agent, monkeypatch): + # Arrange: the model returns a candidate list (extra line trimmed to the cap) + self._set(monkeypatch, forks=3, mode="llm") + agent.client = _mock_client( + _api_response("- Hypothesis A\n- Hypothesis B\n- Hypothesis C\n- Hypothesis D") + ) + + # Act + result = await agent._plan_hypotheses(_anom("latency_spike")) + + # Assert: parsed, de-bulleted, bounded to the cap + assert result == ["Hypothesis A", "Hypothesis B", "Hypothesis C"] + + @pytest.mark.asyncio + async def test_llm_failure_falls_back_to_map(self, agent, monkeypatch): + # Arrange: generation errors out + self._set(monkeypatch, forks=3, mode="llm") + agent.client = _mock_client(anthropic.APITimeoutError(request=None)) + + # Act + result = await agent._plan_hypotheses(_anom("backpressure")) + + # Assert: falls back to the type map, never hinges on the optional pre-step + assert result == ANOMALY_HYPOTHESES["backpressure"][:3] + + @pytest.mark.asyncio + async def test_forks_use_type_specific_hypotheses(self, agent, monkeypatch): + # Arrange + self._set(monkeypatch, forks=3, mode="map") + seen = [] + + async def fake_spawn(anomaly, hypothesis=None): + seen.append(hypothesis) + return _diag() + + monkeypatch.setattr(agent, "_spawn_diagnostic_agent", fake_spawn) + + # Act + result = await agent._run_diagnostics(_anom("throughput_drop")) + + # Assert: each fork seeded with a throughput-drop-specific hypothesis + assert result is not None + assert set(seen) == set(ANOMALY_HYPOTHESES["throughput_drop"][:3]) + + @pytest.mark.asyncio + async def test_fork_count_adapts_to_hypotheses_not_cap(self, agent, monkeypatch): + # Arrange: cap (5) exceeds the mapped hypotheses (3) for this type + self._set(monkeypatch, forks=5, mode="map") + seen = [] + + async def fake_spawn(anomaly, hypothesis=None): + seen.append(hypothesis) + return _diag() + + monkeypatch.setattr(agent, "_spawn_diagnostic_agent", fake_spawn) + + # Act + await agent._run_diagnostics(_anom("latency_spike")) + + # Assert: 3 forks (the hypotheses warranted), not 5 (the cap) + assert len(seen) == 3