From 21007e122f59161e2b0e1ca6c52af3a8c8b9bf16 Mon Sep 17 00:00:00 2001 From: Renn F Date: Thu, 14 May 2026 05:21:33 +0200 Subject: [PATCH] fix(mcp): do_server per-verb circuit breaker mirrors flow_server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Smoke-6 surfaced the gap. main-pm called note(scope='decision') with context: null 8 times in a row — every one returned incomplete_input and the agent kept retrying. The flow_server had a breaker (C1) but do_server didn't, so content-tool rejections went uncapped. Mirror the flow_server pattern: - _CIRCUIT_REJECTION_KINDS = {tracing_gap, invalid_state, not_authorized, incomplete_input} (same set) - _record_and_check_circuit posts to the SDK's /verb/attempted on each counted rejection - When the SDK reports open=true, the original rejection envelope is REPLACED with the circuit_open envelope so the agent stops retrying The SDK side (agent_sdk/server.py) already accepts arbitrary verb names; no changes there. note hits the default cap of 3 retries / 60s from foundation.agent_loop. After the third incomplete_input the agent gets circuit_open and the loop ends. 9 new tests pinning the contract. --- roboco/mcp/do_server.py | 88 ++++++ .../test_do_server_circuit_breaker.py | 262 ++++++++++++++++++ 2 files changed, 350 insertions(+) create mode 100644 tests/unit/mcp_servers/test_do_server_circuit_breaker.py diff --git a/roboco/mcp/do_server.py b/roboco/mcp/do_server.py index 2369d099..c5f05601 100644 --- a/roboco/mcp/do_server.py +++ b/roboco/mcp/do_server.py @@ -27,10 +27,24 @@ ORCHESTRATOR_URL = os.environ.get( "ROBOCO_ORCHESTRATOR_URL", "http://roboco-orchestrator:8000", ) +# Per-agent SDK loopback for the per-verb circuit breaker. +SDK_URL = os.environ.get("ROBOCO_SDK_URL", "http://localhost:9000") AGENT_ID = os.environ["ROBOCO_AGENT_ID"] AGENT_ROLE = os.environ["ROBOCO_AGENT_ROLE"] _TIMEOUT = 30 +# Tight timeout for SDK loopback — local sidecar; gateway path must not stall. +_SDK_TIMEOUT = 2.0 + +# Envelope error kinds that count toward the per-verb circuit breaker. +# Mirrors flow_server._CIRCUIT_REJECTION_KINDS — agent_sdk.server is the +# authoritative side; the same set must be applied here so the do-server +# (content tools) gets the same protection as flow-server (intent verbs). +# Smoke-6 surfaced the gap: `note(scope='decision')` looped 8 times +# returning `incomplete_input` with no breaker. +_CIRCUIT_REJECTION_KINDS: frozenset[str] = frozenset( + {"tracing_gap", "invalid_state", "not_authorized", "incomplete_input"} +) mcp = FastMCP("roboco-do") log = structlog.get_logger() @@ -56,6 +70,13 @@ def _post(path: str, body: dict[str, Any]) -> dict[str, Any]: Mirrors flow_server._post: surfaces the orchestrator's envelope on both 2xx and 4xx so the agent always sees ``remediate``. Only fabricates a transport_error envelope when the body is unparseable. + + Rejection envelopes (error in _CIRCUIT_REJECTION_KINDS) are forwarded + to the local SDK's /verb/attempted so the per-verb circuit breaker + can track them. If the SDK reports open, the original rejection is + REPLACED with circuit_open. Smoke-6 surfaced the gap: do-server had + no breaker and `note(scope='decision')` looped 8 times returning + incomplete_input. """ with httpx.Client(timeout=_TIMEOUT) as client: response = client.post( @@ -78,8 +99,75 @@ def _post(path: str, body: dict[str, Any]) -> dict[str, Any]: ), "missing": [], } + # Outside the orchestrator client so the SDK call is its own connection. + return _record_and_check_circuit(path, body, payload) + + +def _verb_from_path(path: str) -> str: + """Extract the verb name from a do-server path. + + ``/api/v2/do/`` → ````. Returns the original path if it + doesn't match the expected shape (defensive — breaker falls open + downstream when the verb is unrecognized). + """ + return path.rsplit("/", 1)[-1] + + +def _record_and_check_circuit( + path: str, + body: dict[str, Any], + payload: dict[str, Any], +) -> dict[str, Any]: + """Forward a content-tool rejection to the SDK breaker; maybe substitute. + + For successful (ok) envelopes this is a no-op — only rejections of + kind tracing_gap / invalid_state / not_authorized / incomplete_input + are reported. When the SDK responds with ``open=true`` we replace the + original rejection with the wire-format ``circuit_open`` envelope so + the agent stops retrying. + + Best-effort: SDK unreachable, slow, or malformed response → return + the original payload. The breaker is a safety net; it must never + break the gateway path. + """ + rejection_kind = payload.get("error") + if rejection_kind not in _CIRCUIT_REJECTION_KINDS: return payload + verb = _verb_from_path(path) + task_id = body.get("task_id") + try: + with httpx.Client(timeout=_SDK_TIMEOUT) as client: + resp = client.post( + f"{SDK_URL}/verb/attempted", + json={ + "verb": verb, + "task_id": str(task_id) if task_id is not None else None, + "rejection_kind": rejection_kind, + }, + ) + status = resp.json() + except (httpx.HTTPError, OSError, ValueError, json.JSONDecodeError) as exc: + log.warning( + "do_server: SDK /verb/attempted unreachable; breaker bypassed", + verb=verb, + task_id=task_id, + error=str(exc), + ) + return payload + + if status.get("open") and isinstance(status.get("circuit_envelope"), dict): + circuit_env: dict[str, Any] = status["circuit_envelope"] + log.info( + "do_server: circuit_open substituted for rejection", + verb=verb, + task_id=task_id, + attempts=status.get("attempts"), + limit=status.get("limit"), + ) + return circuit_env + return payload + def commit(message: str, files: list[str] | None = None) -> dict[str, Any]: """Make a git commit. [task-id] prefix auto-applied. Validates message.""" diff --git a/tests/unit/mcp_servers/test_do_server_circuit_breaker.py b/tests/unit/mcp_servers/test_do_server_circuit_breaker.py new file mode 100644 index 00000000..0da46aa7 --- /dev/null +++ b/tests/unit/mcp_servers/test_do_server_circuit_breaker.py @@ -0,0 +1,262 @@ +"""do_server wires content-tool rejections into the SDK per-verb circuit breaker. + +Smoke-6 (2026-05-14): be-pm's main-pm hit `note(scope='decision')` with +`context: null` 8 times in a row, each returning `incomplete_input`, and +the agent kept retrying. The breaker existed on flow_server but not on +do_server — content tools had no safety net. + +These tests mirror test_flow_server_circuit_breaker.py: stub httpx.Client +so the orchestrator URL returns the rejection envelope and the SDK URL +returns the breaker state. The do_server's _post must call /verb/attempted +on rejection kinds (tracing_gap, invalid_state, not_authorized, +incomplete_input) and substitute circuit_open when the SDK reports open. +""" + +from __future__ import annotations + +import importlib +import json +from typing import TYPE_CHECKING, Any +from unittest.mock import MagicMock, patch + +import pytest + +if TYPE_CHECKING: + import types + from pathlib import Path + + +_FULL_MANIFEST = { + "agent_id": "00000000-0000-0000-0000-000000000099", + "role": "developer", + "team": "backend", + "workspace_path": "/tmp/test", + "flow_tools": [], + "do_tools": [ + "commit", + "note", + "say", + "dm", + "evidence", + "progress", + "notify", + "open_session", + "link_session", + "notify_list", + "notify_get", + "notify_ack", + "channels", + "pr_update", + ], + "read_tools": ["Read", "Glob", "Grep"], + "write_tools": ["Edit", "Write"], + "bash_allowed": True, + "subagent_allowed": False, + "subagent_model": None, + "env": {}, +} + + +@pytest.fixture() +def do_module(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> types.ModuleType: + """Import do_server with a tmp manifest + known orchestrator/SDK URLs.""" + manifest_path = tmp_path / "tool-manifest.json" + manifest_path.write_text(json.dumps(_FULL_MANIFEST)) + + monkeypatch.setenv("ROBOCO_AGENT_ID", "00000000-0000-0000-0000-000000000099") + monkeypatch.setenv("ROBOCO_AGENT_ROLE", "developer") + monkeypatch.setenv("ROBOCO_ORCHESTRATOR_URL", "http://test-orchestrator:8000") + monkeypatch.setenv("ROBOCO_SDK_URL", "http://test-sdk:9000") + monkeypatch.setenv("ROBOCO_TOOL_MANIFEST_PATH", str(manifest_path)) + + import roboco.mcp.do_server as srv + + importlib.reload(srv) + return srv + + +def _make_client( + orchestrator_response: dict[str, Any], sdk_response: dict[str, Any] | None +): + """Build an httpx.Client mock that dispatches by destination URL.""" + captured: list[tuple[str, dict[str, Any] | None]] = [] + + def _client_factory(*_args: Any, **_kwargs: Any) -> MagicMock: + client = MagicMock() + client.__enter__ = MagicMock(return_value=client) + client.__exit__ = MagicMock(return_value=False) + + def _post(url: str, **kwargs: Any) -> MagicMock: + captured.append((url, kwargs.get("json"))) + resp = MagicMock() + if "test-sdk" in url: + if sdk_response is None: + raise AssertionError("SDK called unexpectedly") + resp.json.return_value = sdk_response + else: + resp.json.return_value = orchestrator_response + return resp + + client.post.side_effect = _post + return client + + return _client_factory, captured + + +def test_ok_envelope_does_not_touch_sdk(do_module: types.ModuleType) -> None: + """An envelope with error=None never POSTs to /verb/attempted.""" + factory, captured = _make_client( + orchestrator_response={ + "status": "noted", + "task_id": None, + "next": "continue", + "error": None, + }, + sdk_response=None, + ) + with patch("httpx.Client", side_effect=factory): + result = do_module.note(text="hi", scope="note") + assert result["error"] is None + assert all("test-sdk" not in url for url, _ in captured) + + +def test_incomplete_input_forwards_to_sdk(do_module: types.ModuleType) -> None: + """A note rejection (incomplete_input) triggers POST /verb/attempted.""" + factory, captured = _make_client( + orchestrator_response={ + "error": "incomplete_input", + "missing": ["context", "chosen", "rationale"], + "remediate": "re-issue note(scope='decision', ...)", + }, + sdk_response={ + "verb": "note", + "task_id": None, + "attempts": 1, + "limit": 3, + "window_seconds": 60, + "open": False, + "circuit_envelope": None, + }, + ) + with patch("httpx.Client", side_effect=factory): + result = do_module.note(text="...", scope="decision") + # Original rejection survives — breaker is not yet open. + assert result["error"] == "incomplete_input" + sdk_calls = [(url, body) for url, body in captured if "test-sdk" in url] + assert len(sdk_calls) == 1 + sdk_url, sdk_body = sdk_calls[0] + assert sdk_url.endswith("/verb/attempted") + assert sdk_body == { + "verb": "note", + "task_id": None, + "rejection_kind": "incomplete_input", + } + + +@pytest.mark.parametrize( + "rejection_kind", + ["tracing_gap", "invalid_state", "not_authorized", "incomplete_input"], +) +def test_all_counted_rejection_kinds_forwarded( + do_module: types.ModuleType, rejection_kind: str +) -> None: + """All four counted error kinds forward to the SDK from do_server.""" + factory, captured = _make_client( + orchestrator_response={ + "error": rejection_kind, + "message": "no", + "remediate": "fix", + }, + sdk_response={ + "verb": "note", + "task_id": None, + "attempts": 1, + "limit": 3, + "window_seconds": 60, + "open": False, + "circuit_envelope": None, + }, + ) + with patch("httpx.Client", side_effect=factory): + do_module.note(text="x") + sdk_calls = [(url, body) for url, body in captured if "test-sdk" in url] + assert len(sdk_calls) == 1 + assert sdk_calls[0][1]["rejection_kind"] == rejection_kind + + +def test_breaker_open_substitutes_circuit_open(do_module: types.ModuleType) -> None: + """When SDK reports open=true, return the circuit_envelope to the agent.""" + circuit_env = { + "error": "circuit_open", + "message": "verb 'note' rejected too often (3 in 60s)", + "remediate": "fix the missing fields once, then retry", + "missing": [], + } + factory, _ = _make_client( + orchestrator_response={ + "error": "incomplete_input", + "missing": ["context"], + "remediate": "fill context", + }, + sdk_response={ + "verb": "note", + "task_id": None, + "attempts": 3, + "limit": 3, + "window_seconds": 60, + "open": True, + "circuit_envelope": circuit_env, + }, + ) + with patch("httpx.Client", side_effect=factory): + result = do_module.note(text="x", scope="decision") + assert result == circuit_env + + +def test_sdk_unreachable_falls_open(do_module: types.ModuleType) -> None: + """If SDK loopback dies, return the original rejection — never break the path.""" + import httpx + + def _client_factory(*_args: Any, **_kwargs: Any) -> MagicMock: + client = MagicMock() + client.__enter__ = MagicMock(return_value=client) + client.__exit__ = MagicMock(return_value=False) + + def _post(url: str, **_kw: Any) -> MagicMock: + if "test-sdk" in url: + raise httpx.ConnectError("connection refused") + resp = MagicMock() + resp.json.return_value = { + "error": "incomplete_input", + "missing": ["context"], + "remediate": "fill context", + } + return resp + + client.post.side_effect = _post + return client + + with patch("httpx.Client", side_effect=_client_factory): + result = do_module.note(text="x", scope="decision") + # Original rejection — never circuit_open when the SDK is unreachable. + assert result["error"] == "incomplete_input" + + +def test_verb_extracted_from_path(do_module: types.ModuleType) -> None: + """commit() reports verb='commit', say() reports verb='say' etc.""" + factory, captured = _make_client( + orchestrator_response={"error": "invalid_state", "message": "no commits"}, + sdk_response={ + "verb": "commit", + "task_id": None, + "attempts": 1, + "limit": 3, + "window_seconds": 60, + "open": False, + "circuit_envelope": None, + }, + ) + with patch("httpx.Client", side_effect=factory): + do_module.commit(message="[abc12345] test commit message under 20 chars") + sdk_body = next(body for url, body in captured if "test-sdk" in url) + assert sdk_body["verb"] == "commit"