feat(agent): dynamic, anomaly-derived diagnostic hypotheses + adaptive fork count

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-<primary>-<index>` 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.

mcp-server/src/streamops_mcp/agent/monitor.py

@@ -49,16 +49,52 @@
 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.",
     "External dependency: a downstream sink, source, or coordination service degrading.",
     "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.

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
 

mcp-server/tests/test_config.py

@@ -1,18 +1,35 @@
 """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"
         assert cfg.prometheus_url == "http://localhost:9090"
         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")