fix(mcp): do_server per-verb circuit breaker mirrors flow_server

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.
2026-08-03 07:23:24 +02:00 · 2026-05-14 05:21:33 +02:00
parent 3171e1e192
commit 21007e122f
2 changed files with 350 additions and 0 deletions
@@ -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/<verb>`` → ``<verb>``. 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."""
@@ -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"