From 7a79ef36118ac70d1cc584731badcc3abeedbe35 Mon Sep 17 00:00:00 2001 From: matt butler Date: Thu, 9 Jul 2026 15:53:14 -0400 Subject: [PATCH] feat(agent): explicit pre-delegation gate before spawning the Report agent Closes #94. Cert Ep 04's fail-fast lesson: a coordinator must verify an upstream sub-agent's structured finding before delegating to the next, and short-circuit if it is thin, or the downstream agent hallucinates on empty context. run_cycle's only gate before the Report agent was _all_claims_low_confidence, which returns False for an empty claims list, so a diagnosis with NO claims (or a parse_error fallback with none) sailed past it into the Report agent, which then synthesized a full incident from nothing. Add _verify_diagnosis_precondition(diagnosis) -> str | None, returning a reason when the diagnosis is not fit to delegate (parse-error fallback, no claims, or all claims LOW/UNSOURCED) and None when it is. run_cycle calls it in place of the inline low-confidence check; on a reason it logs and returns None without spawning the Report agent or escalating. Multi-agent path only; single-agent mode (no sub-agent claims by design) is unchanged. Tests: the gate for reportable / no-claims / parse-error / all-low, plus a run_cycle test asserting an empty diagnosis never spawns the Report agent and never escalates. Full suite 260 passed; ruff + mypy clean. --- mcp-server/src/streamops_mcp/agent/monitor.py | 30 +++++++- mcp-server/tests/test_monitor.py | 74 +++++++++++++++++++ 2 files changed, 100 insertions(+), 4 deletions(-) diff --git a/mcp-server/src/streamops_mcp/agent/monitor.py b/mcp-server/src/streamops_mcp/agent/monitor.py index 00b7383..41e72e5 100644 --- a/mcp-server/src/streamops_mcp/agent/monitor.py +++ b/mcp-server/src/streamops_mcp/agent/monitor.py @@ -193,11 +193,13 @@ async def run_cycle(self) -> IncidentReport | None: self._log_confidence_distribution(diagnosis) - if self._all_claims_low_confidence(diagnosis): + # Fail-fast: verify the diagnosis is substantive before delegating to + # the Report agent, so it never synthesizes an incident from thin or + # degraded context (cert Ep 04 pattern). See issue #94. + gate_reason = self._verify_diagnosis_precondition(diagnosis) + if gate_reason is not None: logger.warning( - "All %d claims are LOW or UNSOURCED confidence; " - "downgrading to warning-level log, skipping report agent", - len(diagnosis.claims), + "Skipping Report Agent, diagnosis not reportable: %s", gate_reason ) return None @@ -504,6 +506,26 @@ def _all_claims_low_confidence(diagnosis: DiagnosisReport) -> bool: low_levels = {Confidence.LOW, Confidence.UNSOURCED} return all(c.confidence in low_levels for c in diagnosis.claims) + def _verify_diagnosis_precondition(self, diagnosis: DiagnosisReport) -> str | None: + """Check the diagnosis is substantive enough to delegate to the Report agent. + + Returns a short reason string when the diagnosis is NOT fit to delegate, + or None when it is. Delegating a thin or degraded finding invites the + downstream Report agent to synthesize an incident from nothing (cert Ep 04 + fail-fast delegation). Gated cases: + - a parse-error fallback diagnosis (unstructured, produced when the + Diagnostic agent's output could not be parsed), + - a diagnosis with no claims (no structured finding to report on), + - all claims LOW or UNSOURCED (no confident finding). + """ + if diagnosis.anomaly_type == "parse_error": + return "diagnosis is a parse-error fallback (unstructured)" + if not diagnosis.claims: + return "diagnosis has no claims to report on" + if self._all_claims_low_confidence(diagnosis): + return f"all {len(diagnosis.claims)} claims are LOW or UNSOURCED confidence" + return None + @staticmethod def _extract_low_confidence_claims(diagnosis: DiagnosisReport) -> list[str]: """Extract claim text for LOW and UNSOURCED claims.""" diff --git a/mcp-server/tests/test_monitor.py b/mcp-server/tests/test_monitor.py index 137793e..d2921cd 100644 --- a/mcp-server/tests/test_monitor.py +++ b/mcp-server/tests/test_monitor.py @@ -1195,3 +1195,77 @@ async def fake_spawn(anomaly, hypothesis=None): # Assert: 3 forks (the hypotheses warranted), not 5 (the cap) assert len(seen) == 3 + + +def _empty_diagnosis() -> DiagnosisReport: + """A structurally valid diagnosis the Diagnostic agent produced with no claims.""" + return DiagnosisReport( + anomaly_type="latency_spike", + detected_at="2026-07-09T12:00:00Z", + sources=[], + claims=[], + affected_components=[], + root_cause=RootCause(summary="inconclusive", confidence="low", reasoning="none"), + tools_used=[], + ) + + +class TestPreDelegationGate: + """The coordinator must not delegate a thin/degraded finding to the Report agent (Ep 04, issue #94).""" + + def test_reportable_diagnosis_passes(self, agent): + # Arrange: a diagnosis with a HIGH-confidence claim + # Act + reason = agent._verify_diagnosis_precondition(_diag(confidence="HIGH")) + + # Assert + assert reason is None + + def test_no_claims_is_gated(self, agent): + # Arrange + Act + reason = agent._verify_diagnosis_precondition(_empty_diagnosis()) + + # Assert + assert reason is not None + assert "no claims" in reason + + def test_parse_error_fallback_is_gated(self, agent): + # Arrange: the parse-error fallback shape (even if it somehow had a claim) + diagnosis = _diag(confidence="HIGH") + diagnosis.anomaly_type = "parse_error" + + # Act + reason = agent._verify_diagnosis_precondition(diagnosis) + + # Assert + assert reason is not None + assert "parse-error" in reason + + def test_all_low_confidence_is_gated(self, agent): + # Arrange + # Act + reason = agent._verify_diagnosis_precondition(_diag(confidence="LOW")) + + # Assert + assert reason is not None + assert "LOW or UNSOURCED" in reason + + @pytest.mark.asyncio + async def test_run_cycle_does_not_delegate_empty_diagnosis(self, agent, monkeypatch): + # Arrange: detection fires, but the Diagnostic agent returns no claims + monkeypatch.setattr("streamops_mcp.agent.monitor.config.agent_diagnostic_forks", 1) + monkeypatch.setattr( + agent, "_detect_anomalies", AsyncMock(return_value=_anom("latency_spike")) + ) + monkeypatch.setattr(agent, "_run_diagnostics", AsyncMock(return_value=_empty_diagnosis())) + spawn_report = AsyncMock() + monkeypatch.setattr(agent, "_spawn_report_agent", spawn_report) + + # Act + with patch("streamops_mcp.agent.monitor.escalate", new_callable=AsyncMock) as mock_escalate: + result = await agent.run_cycle() + + # Assert: fail-fast, the Report agent is never spawned and nothing escalates + assert result is None + spawn_report.assert_not_awaited() + mock_escalate.assert_not_awaited()