From c36fd385de19364cd8575afd0090225ce9ebaa51 Mon Sep 17 00:00:00 2001 From: matt butler Date: Thu, 9 Jul 2026 15:07:20 -0400 Subject: [PATCH] feat(agent): fork-style parallel diagnostic exploration Closes #67. One diagnostic agent is a single line of reasoning, prone to tunnel vision on an ambiguous anomaly with several plausible causes. The coordinator can now fan out N diagnostic sub-agents concurrently from the shared detection baseline, each seeded with a distinct hypothesis, then merge the survivors. - config.agent_diagnostic_forks (default 1) gates fan-out; >1 forks (capped at the number of defined hypotheses). Default preserves single-agent behavior. - _run_diagnostics: single-agent path unchanged; fork path spawns via asyncio.gather (truly concurrent, not a loop), each fork retried independently (#51). Partial failure tolerated: survivors aggregated; abort only if every fork fails. - _merge_fork_diagnoses: namespaces each fork's ids (f{i}:) so merged sources, claims, and conflicts never collide; attribution preserved (the merged report re-validates). Forks that concluded a different anomaly_type than the primary (highest-confidence) fork produce a cross-fork ConflictRecord: unresolved, escalated via the existing #82 path, never a silent winner. anomaly_type is the deterministic disagreement signal; semantic claim-level reconciliation would need an LLM (noted out of scope). - _spawn_diagnostic_agent takes an optional hypothesis seed injected into the prompt. Tests: single-agent path when forks=1, concurrent spawn with distinct hypotheses (asyncio.Barrier proves concurrency, not sequential), partial-failure aggregation, all-fail abort, merge namespacing + attribution, cross-fork conflict creation, and the fork path end-to-end through run_cycle. Full suite 244 passed; ruff + mypy clean. --- mcp-server/src/streamops_mcp/agent/monitor.py | 214 ++++++++++++++++-- mcp-server/src/streamops_mcp/config.py | 6 + mcp-server/tests/test_monitor.py | 180 +++++++++++++++ 3 files changed, 386 insertions(+), 14 deletions(-) diff --git a/mcp-server/src/streamops_mcp/agent/monitor.py b/mcp-server/src/streamops_mcp/agent/monitor.py index 3c84908..ea67fa6 100644 --- a/mcp-server/src/streamops_mcp/agent/monitor.py +++ b/mcp-server/src/streamops_mcp/agent/monitor.py @@ -25,6 +25,7 @@ from streamops_mcp.agent.executor import execute_tool from streamops_mcp.agent.schemas import ( Confidence, + ConflictRecord, DetectedAnomaly, DiagnosisReport, DiagnosticToReportHandoff, @@ -48,6 +49,24 @@ 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. +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.", +] + +# Confidence ordering for picking a fork's representative claim and the primary fork. +_CONFIDENCE_RANK = { + Confidence.HIGH: 3, + Confidence.MEDIUM: 2, + Confidence.LOW: 1, + Confidence.UNSOURCED: 0, +} + class MonitorAgent: """Coordinator agent that runs the monitoring loop. @@ -132,16 +151,8 @@ async def run_cycle(self) -> IncidentReport | None: return None if self.multi_agent: - try: - diagnosis = await self._retry_subagent( - "Diagnostic Agent", - lambda: self._spawn_diagnostic_agent(detection), - ) - except Exception as exc: - logger.error( - "Diagnostic Agent failed after retries, cycle aborted: %s", - exc, - ) + diagnosis = await self._run_diagnostics(detection) + if diagnosis is None: return None self._log_confidence_distribution(diagnosis) @@ -193,6 +204,170 @@ async def run_cycle(self) -> IncidentReport | None: finally: 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). + """ + fork_count = min(config.agent_diagnostic_forks, len(DIAGNOSTIC_HYPOTHESES)) + + if fork_count <= 1: + try: + return await self._retry_subagent( + "Diagnostic Agent", + lambda: self._spawn_diagnostic_agent(anomaly), + ) + except Exception as exc: + logger.error("Diagnostic Agent failed after retries, cycle aborted: %s", exc) + return None + + hypotheses = DIAGNOSTIC_HYPOTHESES[:fork_count] + logger.info("Fanning out %d diagnostic forks", fork_count) + + async def _fork(index: int, hypothesis: str) -> DiagnosisReport: + return await self._retry_subagent( + f"Diagnostic Agent fork {index}", + lambda: self._spawn_diagnostic_agent(anomaly, hypothesis=hypothesis), + ) + + # gather (not a loop) so the forks truly run concurrently on a shared baseline. + results = await asyncio.gather( + *(_fork(i, h) for i, h in enumerate(hypotheses)), + return_exceptions=True, + ) + survivors = [r for r in results if isinstance(r, DiagnosisReport)] + failed = fork_count - len(survivors) + + if not survivors: + logger.error("All %d diagnostic forks failed after retries, cycle aborted", fork_count) + return None + if failed: + logger.warning( + "%d of %d diagnostic forks failed; aggregating %d survivor(s)", + failed, + fork_count, + len(survivors), + ) + return self._merge_fork_diagnoses(survivors) + + @classmethod + def _merge_fork_diagnoses(cls, diagnoses: list[DiagnosisReport]) -> DiagnosisReport: + """Merge parallel fork diagnoses into one, preserving attribution. + + Each fork's ids are namespaced (``f{i}:``) so merged sources, claims, and + conflicts never collide. Forks that concluded a different anomaly_type + than the primary (highest-confidence) fork produce a cross-fork + ConflictRecord: annotated, unresolved, and left for the coordinator's + escalation path (issue #82) rather than silently picking a winner. + """ + if len(diagnoses) == 1: + return diagnoses[0] + + merged_sources = [] + merged_claims = [] + merged_conflicts = [] + reps: list[tuple[int, str | None, str, str]] = [] # (fork, rep_claim_id, type, summary) + + for i, d in enumerate(diagnoses): + prefix = f"f{i}:" + for s in d.sources: + merged_sources.append(s.model_copy(update={"source_id": prefix + s.source_id})) + for c in d.claims: + merged_claims.append( + c.model_copy( + update={"claim_id": prefix + c.claim_id, "source_id": prefix + c.source_id} + ) + ) + for cf in d.conflicts: + merged_conflicts.append( + cf.model_copy( + update={ + "conflict_id": prefix + cf.conflict_id, + "claim_a_id": prefix + cf.claim_a_id, + "claim_b_id": prefix + cf.claim_b_id, + } + ) + ) + reps.append( + (i, cls._representative_claim_id(d, prefix), d.anomaly_type, d.root_cause.summary) + ) + + primary = cls._primary_fork_index(diagnoses) + merged_conflicts.extend(cls._cross_fork_conflicts(reps, primary)) + + chosen = diagnoses[primary] + return DiagnosisReport( + anomaly_type=chosen.anomaly_type, + detected_at=min(d.detected_at for d in diagnoses), + sources=merged_sources, + claims=merged_claims, + conflicts=merged_conflicts, + affected_components=[c for d in diagnoses for c in d.affected_components], + root_cause=chosen.root_cause, + tools_used=sorted({t for d in diagnoses for t in d.tools_used}), + raw_evidence=[e for d in diagnoses for e in d.raw_evidence], + ) + + @classmethod + def _representative_claim_id(cls, diagnosis: DiagnosisReport, prefix: str) -> str | None: + """The namespaced id of a fork's highest-confidence claim, or None if it has none.""" + if not diagnosis.claims: + return None + best = max(diagnosis.claims, key=lambda c: _CONFIDENCE_RANK[c.confidence]) + return prefix + best.claim_id + + @classmethod + def _primary_fork_index(cls, diagnoses: list[DiagnosisReport]) -> int: + """Index of the fork whose best claim has the highest confidence (ties: lowest index).""" + + def fork_rank(i: int) -> int: + claims = diagnoses[i].claims + return max((_CONFIDENCE_RANK[c.confidence] for c in claims), default=-1) + + return max(range(len(diagnoses)), key=fork_rank) + + @staticmethod + def _cross_fork_conflicts( + reps: list[tuple[int, str | None, str, str]], primary: int + ) -> list[ConflictRecord]: + """Pair the primary fork against each fork that concluded a different anomaly_type. + + anomaly_type is the deterministic, structured disagreement signal (comparing + free-text root causes would flag every fork as different). Semantic + claim-level reconciliation would need an LLM and is out of scope here. + Only forks that produced at least one claim can be referenced. + """ + by_index = {r[0]: r for r in reps} + p = by_index.get(primary) + if p is None or p[1] is None: + return [] + _, p_claim, p_type, p_summary = p + assert p_claim is not None # guarded by p[1] is None check above + + conflicts = [] + for index, rep_claim, anomaly_type, summary in reps: + if index == primary or rep_claim is None: + continue + if anomaly_type != p_type: + conflicts.append( + ConflictRecord( + conflict_id=f"xf-{primary}-{index}", + topic="cross-fork root-cause disagreement", + claim_a_id=p_claim, + claim_b_id=rep_claim, + resolution="unresolved", + notes=( + f"Fork {primary} concluded '{p_type}' ({p_summary}); " + f"fork {index} concluded '{anomaly_type}' ({summary})" + ), + ) + ) + return conflicts + @staticmethod def _fallback_report(diagnosis: DiagnosisReport) -> IncidentReport: """Produce a fallback IncidentReport when the Report Agent fails.""" @@ -313,15 +488,21 @@ async def _detect_anomalies(self) -> DetectedAnomaly | None: else None ) - async def _spawn_diagnostic_agent(self, anomaly: DetectedAnomaly) -> DiagnosisReport: + async def _spawn_diagnostic_agent( + self, anomaly: DetectedAnomaly, hypothesis: str | None = None + ) -> DiagnosisReport: """Spawn a Diagnostic sub-agent with scoped context and tools. The sub-agent starts with a blank context. All relevant information must be injected explicitly via the prompt (not inherited from the coordinator's conversation history), as a typed DetectedAnomaly rather - than a prose string. + than a prose string. When a hypothesis is given (fork-style exploration, + issue #67), the agent is steered to investigate that angle first. """ - logger.info("Spawning Diagnostic Agent") + logger.info( + "Spawning Diagnostic Agent%s", + f" (hypothesis: {hypothesis})" if hypothesis else "", + ) schema_hint = DiagnosisReport.model_json_schema() handoff = MonitorToDiagnosticHandoff( @@ -340,13 +521,18 @@ async def _spawn_diagnostic_agent(self, anomaly: DetectedAnomaly) -> DiagnosisRe if runbook_section: system_prompt = system_prompt + "\n\n" + runbook_section + hypothesis_line = ( + f"\nPrioritize investigating this hypothesis before others: {hypothesis}\n" + if hypothesis + else "" + ) messages: list[dict[str, Any]] = [ { "role": "user", "content": f"""Investigate the following anomaly detected by the monitoring system: {anomaly_json} - +{hypothesis_line} Use the available tools to determine the root cause. Respond with a JSON object matching the DiagnosisReport schema: {handoff.schema_hint}""", } diff --git a/mcp-server/src/streamops_mcp/config.py b/mcp-server/src/streamops_mcp/config.py index 4bddc97..37d7b77 100644 --- a/mcp-server/src/streamops_mcp/config.py +++ b/mcp-server/src/streamops_mcp/config.py @@ -60,6 +60,12 @@ 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. + agent_diagnostic_forks: int = 1 + # Input sanitization agent_sanitize_max_output_chars: int = 20_000 diff --git a/mcp-server/tests/test_monitor.py b/mcp-server/tests/test_monitor.py index 7ac40b1..c26a5dd 100644 --- a/mcp-server/tests/test_monitor.py +++ b/mcp-server/tests/test_monitor.py @@ -4,6 +4,7 @@ parsing, retry/fallback behavior) without making actual Claude API calls. """ +import asyncio import json from types import SimpleNamespace from unittest.mock import AsyncMock, patch @@ -16,11 +17,14 @@ from streamops_mcp.agent.schemas import ( ClaimRecord, Confidence, + DetectedAnomaly, DiagnosisReport, IncidentReport, + RootCause, Severity, SourceRecord, ) +from streamops_mcp.agent.schemas.diagnosis import AffectedComponent def _fake_response(status_code: int) -> httpx.Response: @@ -884,3 +888,179 @@ def test_fallback_report_carries_attribution(self, agent): # Assert assert len(report.sources) == len(diagnosis.sources) assert len(report.supporting_claims) == len(diagnosis.claims) + + +def _anom(anomaly_type="unknown") -> DetectedAnomaly: + return DetectedAnomaly( + anomaly_type=anomaly_type, summary="ambiguous anomaly", detected_at="2026-07-09T12:00:00Z" + ) + + +def _diag( + anomaly_type="latency_spike", confidence="HIGH", sid="src-001", cid="C01" +) -> DiagnosisReport: + return DiagnosisReport( + anomaly_type=anomaly_type, + detected_at="2026-07-09T12:00:00Z", + sources=[ + SourceRecord( + source_id=sid, tool_name="query_flink_jobs", retrieved_at="t", raw_output="{}" + ) + ], + claims=[ClaimRecord(claim_id=cid, text="a finding", source_id=sid, confidence=confidence)], + affected_components=[ + AffectedComponent(name="kafka", role="consumer", status="degraded", evidence="lag") + ], + root_cause=RootCause( + summary=f"{anomaly_type} root cause", confidence="medium", reasoning="r" + ), + tools_used=["query_flink_jobs"], + ) + + +class TestForkDiagnostics: + """Fork-style parallel diagnostic exploration (issue #67).""" + + @pytest.mark.asyncio + async def test_single_agent_path_when_forks_one(self, agent, monkeypatch): + # Arrange: default config -> one diagnostic agent, no hypothesis + monkeypatch.setattr("streamops_mcp.agent.monitor.config.agent_diagnostic_forks", 1) + seen = {"count": 0, "hypothesis": "sentinel"} + + async def fake_spawn(anomaly, hypothesis=None): + seen["count"] += 1 + seen["hypothesis"] = hypothesis + return _diag() + + monkeypatch.setattr(agent, "_spawn_diagnostic_agent", fake_spawn) + + # Act + result = await agent._run_diagnostics(_anom()) + + # Assert + assert result is not None + assert seen["count"] == 1 + assert seen["hypothesis"] is None + + @pytest.mark.asyncio + async def test_forks_run_concurrently_with_distinct_hypotheses(self, agent, monkeypatch): + # Arrange: 3 forks; a barrier proves they run concurrently, not sequentially + monkeypatch.setattr("streamops_mcp.agent.monitor.config.agent_diagnostic_forks", 3) + barrier = asyncio.Barrier(3) + seen_hypotheses = [] + + async def fake_spawn(anomaly, hypothesis=None): + seen_hypotheses.append(hypothesis) + # If forks were sequential, only 1 of 3 would reach the barrier and + # this wait would time out. Concurrent execution releases it at once. + await asyncio.wait_for(barrier.wait(), timeout=2.0) + return _diag() + + monkeypatch.setattr(agent, "_spawn_diagnostic_agent", fake_spawn) + + # Act + result = await agent._run_diagnostics(_anom()) + + # Assert + assert result is not None + assert len(seen_hypotheses) == 3 + assert len(set(seen_hypotheses)) == 3 # each fork got a distinct hypothesis + + @pytest.mark.asyncio + async def test_partial_fork_failure_aggregates_survivors(self, agent, monkeypatch): + # Arrange: 3 forks, one dies after retries + monkeypatch.setattr("streamops_mcp.agent.monitor.config.agent_diagnostic_forks", 3) + monkeypatch.setattr("streamops_mcp.agent.monitor.config.agent_max_retries", 0) + + async def fake_spawn(anomaly, hypothesis=None): + if hypothesis and "Data-side" in hypothesis: + raise anthropic.APITimeoutError(request=None) + return _diag() + + monkeypatch.setattr(agent, "_spawn_diagnostic_agent", fake_spawn) + + # Act + result = await agent._run_diagnostics(_anom()) + + # Assert: survivors aggregated, not aborted + assert result is not None + assert len(result.claims) == 2 # two forks survived, one per surviving fork + + @pytest.mark.asyncio + async def test_all_forks_fail_aborts_cycle(self, agent, monkeypatch): + # Arrange: every fork dies + monkeypatch.setattr("streamops_mcp.agent.monitor.config.agent_diagnostic_forks", 3) + monkeypatch.setattr("streamops_mcp.agent.monitor.config.agent_max_retries", 0) + + async def fake_spawn(anomaly, hypothesis=None): + raise anthropic.APITimeoutError(request=None) + + monkeypatch.setattr(agent, "_spawn_diagnostic_agent", fake_spawn) + + # Act + result = await agent._run_diagnostics(_anom()) + + # Assert + assert result is None + + def test_merge_single_returns_input(self): + # Arrange + d = _diag() + + # Act + Assert: nothing to merge + assert MonitorAgent._merge_fork_diagnoses([d]) is d + + def test_merge_namespaces_ids_and_preserves_attribution(self): + # Arrange: two forks that agree + d1 = _diag(anomaly_type="latency_spike") + d2 = _diag(anomaly_type="latency_spike") + + # Act + merged = MonitorAgent._merge_fork_diagnoses([d1, d2]) + + # Assert: ids namespaced per fork (no collision), attribution intact (construction validated) + assert {s.source_id for s in merged.sources} == {"f0:src-001", "f1:src-001"} + assert {c.claim_id for c in merged.claims} == {"f0:C01", "f1:C01"} + for claim in merged.claims: + assert claim.source_id in {s.source_id for s in merged.sources} + # agreement -> no cross-fork conflict + assert not [c for c in merged.conflicts if c.topic == "cross-fork root-cause disagreement"] + + def test_merge_creates_cross_fork_conflict_on_disagreement(self): + # Arrange: forks disagree on anomaly_type; fork 0 is HIGH (primary), fork 1 MEDIUM + d1 = _diag(anomaly_type="latency_spike", confidence="HIGH") + d2 = _diag(anomaly_type="backpressure", confidence="MEDIUM") + + # Act + merged = MonitorAgent._merge_fork_diagnoses([d1, d2]) + + # Assert: a cross-fork conflict, unresolved, referencing real merged claims + xf = [c for c in merged.conflicts if c.topic == "cross-fork root-cause disagreement"] + assert len(xf) == 1 + assert xf[0].resolution == "unresolved" + claim_ids = {c.claim_id for c in merged.claims} + assert xf[0].claim_a_id in claim_ids + assert xf[0].claim_b_id in claim_ids + # primary (highest-confidence) fork drives the merged headline + assert merged.anomaly_type == "latency_spike" + + @pytest.mark.asyncio + async def test_run_cycle_fork_path_end_to_end(self, agent, monkeypatch): + # Arrange: 3 forks through the full cycle (detect -> 3 diagnostics -> report) + monkeypatch.setattr("streamops_mcp.agent.monitor.config.agent_diagnostic_forks", 3) + agent.client = _mock_client( + _api_response("Anomaly detected: consumer lag spike on partition 2"), + _api_response(_DIAGNOSIS_JSON), + _api_response(_DIAGNOSIS_JSON), + _api_response(_DIAGNOSIS_JSON), + _api_response(_REPORT_JSON), + ) + + # Act + with patch("streamops_mcp.agent.monitor.escalate", new_callable=AsyncMock) as mock_escalate: + result = await agent.run_cycle() + + # Assert: 1 detect + 3 concurrent forks + 1 report, escalated once + assert isinstance(result, IncidentReport) + assert agent.client.messages.create.call_count == 5 + mock_escalate.assert_awaited_once()