mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
[chore] mcp-servers: normalize exception bodies to Envelope + lift task_id/correlation_id on circuit_open (#232 #359 #57)
flow_server/do_server: the non-404 JSON path returned exception-handler bodies raw (dict `error` from roboco/generic/http exception handlers, or a 422 `detail` list) — neither is the Envelope wire format the agent is prompted to trust (string error kind + message + remediate + missing), so on any service/validation failure the agent got no remediate and flailed until the breaker tripped. _normalize_exception_envelope lifts the body into a real Envelope (code -> counted string kind via _classify_dict_error_code, NOT_FOUND -> not_found, message lifted, remediate synthesized, missing=[]; 422 -> incomplete_input with the validation detail preserved). The synthesized envelope still flows through the breaker so a 500/422 storm trips it. _record_and_check_circuit: the circuit_open substitution dropped task_id / correlation_id from the top level (the SDK's envelope omits them); lift them from the original rejection so the agent's envelope contract and ops audit-join of the trip event still work, not just nested in inner. intake_server._post_event: capture the relay response body under `detail` on non-success so the grok intake agent gets the real reason (e.g. 'session not in MegaTask scope' on a 422) instead of an opaque http_422 token with no remediation. TDD red->green; ruff + mypy clean; 157 mcp/SDK-breaker tests pass.
This commit is contained in:
Renn F
@@ -273,13 +273,15 @@ def test_verb_extracted_from_path(do_module: types.ModuleType) -> None:
|
||||
assert sdk_body["verb"] == "commit"
|
||||
|
||||
|
||||
def test_dict_shaped_error_does_not_crash(do_module: types.ModuleType) -> None:
|
||||
def test_dict_shaped_error_normalized_to_envelope(do_module: types.ModuleType) -> None:
|
||||
"""A RobocoError.to_dict()-shaped response must not TypeError the breaker.
|
||||
|
||||
A dict-shaped `error` is a retry-storm-worthy rejection (the orchestrator's
|
||||
exception handlers surface this shape on 4xx/5xx), so the breaker must count
|
||||
it via the classifier rather than passing it through silently. The original
|
||||
dict payload still reaches the agent (the breaker only substitutes when open).
|
||||
it via the classifier. The dict body is normalized to an Envelope wire format
|
||||
(#232): the agent receives a string `error` kind (not the dict, which violates
|
||||
the Envelope contract) with the message lifted and a remediate. The breaker
|
||||
still counts it (the synthesized string kind is in the counted set).
|
||||
"""
|
||||
factory, captured = _make_client(
|
||||
orchestrator_response={
|
||||
@@ -299,13 +301,16 @@ def test_dict_shaped_error_does_not_crash(do_module: types.ModuleType) -> None:
|
||||
"circuit_envelope": None,
|
||||
},
|
||||
)
|
||||
# No TypeError; the dict-shaped rejection is forwarded to the SDK.
|
||||
# No TypeError; the dict-shaped rejection is normalized + forwarded to the SDK.
|
||||
with patch("httpx.Client", side_effect=factory):
|
||||
result = do_module.dm(recipient="qa-all", text="x")
|
||||
# Original dict payload still reaches the agent (breaker not open).
|
||||
assert isinstance(result["error"], dict)
|
||||
assert result["error"]["code"] == "A2A_ACCESS_DENIED"
|
||||
# SDK breaker MUST now be called so a storm of these counts.
|
||||
# Normalized to a string-kind Envelope — the dict `error` never reaches the
|
||||
# agent (#232). A2A_ACCESS_DENIED maps to not_authorized (exact-code, #161).
|
||||
assert result["error"] == "not_authorized"
|
||||
assert result["message"] == "be-qa cannot A2A with qa-all"
|
||||
assert isinstance(result["remediate"], str) and result["remediate"]
|
||||
assert result["missing"] == []
|
||||
# SDK breaker MUST still be called so a storm of these counts.
|
||||
sdk_calls = [(url, body) for url, body in captured if "test-sdk" in url]
|
||||
assert len(sdk_calls) == 1
|
||||
# ACCESS_DENIED maps to the not_authorized counted kind.
|
||||
@@ -460,3 +465,123 @@ def test_missing_route_404_returns_envelope_and_counts(
|
||||
_, sdk_body = sdk_calls[0]
|
||||
assert sdk_body is not None
|
||||
assert sdk_body["rejection_kind"] == "invalid_state"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# #232: a non-404 JSON exception-handler body (dict `error` / 422 `detail`)
|
||||
# must be normalized to an Envelope wire format before reaching the agent.
|
||||
# Mirrors flow_server.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_422_detail_normalized_to_incomplete_input_envelope(
|
||||
do_module: types.ModuleType,
|
||||
) -> None:
|
||||
"""A 422 body (`{"detail": [...]}`, no `error`) is normalized to an Envelope
|
||||
with `error='incomplete_input'`, a non-empty `remediate`, `missing=[]`, and
|
||||
the raw validation `detail` preserved. The breaker still counts it. Mirrors
|
||||
flow_server (#232)."""
|
||||
detail_body = [
|
||||
{"loc": ["body", "text"], "msg": "field required", "type": "missing"}
|
||||
]
|
||||
factory, captured = _make_client(
|
||||
orchestrator_response={"detail": detail_body, "body": None},
|
||||
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="")
|
||||
assert result["error"] == "incomplete_input"
|
||||
assert isinstance(result["remediate"], str) and result["remediate"]
|
||||
assert result["missing"] == []
|
||||
assert result["detail"] == detail_body
|
||||
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"] == "incomplete_input"
|
||||
|
||||
|
||||
def test_dict_internal_error_normalized_to_invalid_state_envelope(
|
||||
do_module: types.ModuleType,
|
||||
) -> None:
|
||||
"""A dict-shaped INTERNAL_ERROR (generic_exception_handler) is normalized to
|
||||
`error='invalid_state'` with the message lifted + a remediate. Mirrors
|
||||
flow_server (#232)."""
|
||||
factory, captured = _make_client(
|
||||
orchestrator_response={
|
||||
"error": {
|
||||
"code": "INTERNAL_ERROR",
|
||||
"message": "An internal error occurred",
|
||||
"details": {"correlation_id": "abc"},
|
||||
}
|
||||
},
|
||||
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):
|
||||
result = do_module.commit(message="[abc12345] a valid commit message here")
|
||||
assert result["error"] == "invalid_state"
|
||||
assert result["message"] == "An internal error occurred"
|
||||
assert isinstance(result["remediate"], str) and result["remediate"]
|
||||
assert result["missing"] == []
|
||||
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"] == "invalid_state"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# #359: the circuit_open substitution lifts task_id/correlation_id from the
|
||||
# original rejection to the top-level envelope. Mirrors flow_server.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_circuit_open_lifts_task_id_and_correlation_id(
|
||||
do_module: types.ModuleType,
|
||||
) -> None:
|
||||
"""The SDK's circuit_envelope omits task_id/correlation_id; the substitution
|
||||
lifts them from the original rejection to the top level so the agent's
|
||||
envelope contract and ops audit-join still work. Mirrors flow_server (#359)."""
|
||||
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",
|
||||
"task_id": "T7",
|
||||
"correlation_id": "C7",
|
||||
},
|
||||
sdk_response={
|
||||
"verb": "note",
|
||||
"task_id": "T7",
|
||||
"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["error"] == "circuit_open"
|
||||
assert result["task_id"] == "T7"
|
||||
assert result["correlation_id"] == "C7"
|
||||
assert result["inner"]["error"] == "incomplete_input"
|
||||
assert result["inner"]["task_id"] == "T7"
|
||||
|
||||
@@ -547,11 +547,13 @@ def test_dict_shaped_internal_error_counts_as_invalid_state(
|
||||
assert sdk_calls[0][1]["rejection_kind"] == "invalid_state"
|
||||
|
||||
|
||||
def test_dict_shaped_not_found_does_not_count(
|
||||
def test_dict_shaped_not_found_normalized_to_envelope_and_not_counted(
|
||||
flow_module: types.ModuleType,
|
||||
) -> None:
|
||||
"""A dict-shaped NOT_FOUND (404 family) does NOT count — parity with the
|
||||
string-error contract that a `not_found` rejection isn't counted.
|
||||
"""A dict-shaped NOT_FOUND (404 family) is normalized to an Envelope
|
||||
(`error="not_found"` string + lifted message + remediate) — the agent
|
||||
never sees a dict `error`, which violates the Envelope wire format — and
|
||||
the breaker is NOT touched (parity with the string `not_found` contract).
|
||||
"""
|
||||
factory, captured = _make_client(
|
||||
orchestrator_response={
|
||||
@@ -561,8 +563,13 @@ def test_dict_shaped_not_found_does_not_count(
|
||||
)
|
||||
with patch("httpx.Client", side_effect=factory):
|
||||
result = flow_module.i_am_done("task-A")
|
||||
assert isinstance(result["error"], dict)
|
||||
assert result["error"]["code"] == "TASK_NOT_FOUND"
|
||||
# Normalized to a string-kind Envelope — the dict `error` never reaches
|
||||
# the agent (#232).
|
||||
assert result["error"] == "not_found"
|
||||
assert result["message"] == "no such task"
|
||||
assert isinstance(result["remediate"], str) and result["remediate"]
|
||||
assert result["missing"] == []
|
||||
# NOT_FOUND still does not count toward the breaker.
|
||||
assert all("test-sdk" not in url for url, _ in captured)
|
||||
|
||||
|
||||
@@ -630,3 +637,165 @@ def test_missing_route_404_returns_envelope_and_counts(
|
||||
_, sdk_body = sdk_calls[0]
|
||||
assert sdk_body is not None
|
||||
assert sdk_body["rejection_kind"] == "invalid_state"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# #232: a non-404 JSON exception-handler body (dict `error` / 422 `detail`)
|
||||
# must be normalized to an Envelope wire format before reaching the agent.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_exception_dict_500_normalized_to_invalid_state_envelope(
|
||||
flow_module: types.ModuleType,
|
||||
) -> None:
|
||||
"""A 500 with a dict-shaped error (generic_exception_handler) is normalized
|
||||
to an Envelope: top-level `error='invalid_state'` (string kind), the dict's
|
||||
`message` lifted to top-level `message`, a non-empty `remediate`, `missing=[]`.
|
||||
The agent never receives a dict `error` (which violates the Envelope contract
|
||||
it's prompted to trust). The breaker still counts it as `invalid_state`."""
|
||||
factory, captured = _make_client(
|
||||
orchestrator_response={
|
||||
"error": {
|
||||
"code": "INTERNAL_ERROR",
|
||||
"message": "An internal error occurred",
|
||||
"details": {"correlation_id": "abc"},
|
||||
}
|
||||
},
|
||||
sdk_response={
|
||||
"verb": "i_am_done",
|
||||
"task_id": "task-A",
|
||||
"attempts": 1,
|
||||
"limit": 3,
|
||||
"window_seconds": 60,
|
||||
"open": False,
|
||||
"circuit_envelope": None,
|
||||
},
|
||||
)
|
||||
with patch("httpx.Client", side_effect=factory):
|
||||
result = flow_module.i_am_done("task-A")
|
||||
assert result["error"] == "invalid_state"
|
||||
assert result["message"] == "An internal error occurred"
|
||||
assert isinstance(result["remediate"], str) and result["remediate"]
|
||||
assert result["missing"] == []
|
||||
# The synthesized Envelope still flows through the breaker (string kind in
|
||||
# the counted set) so a 500 storm trips it.
|
||||
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"] == "invalid_state"
|
||||
|
||||
|
||||
def test_422_detail_normalized_to_incomplete_input_envelope(
|
||||
flow_module: types.ModuleType,
|
||||
) -> None:
|
||||
"""A 422 request-validation body (`{"detail": [...]}`, no `error`) is
|
||||
normalized to an Envelope with `error='incomplete_input'`, a non-empty
|
||||
`remediate`, and `missing=[]`; the raw validation `detail` is preserved so
|
||||
the agent can see WHICH fields failed. The breaker still counts it."""
|
||||
detail_body = [
|
||||
{"loc": ["body", "task_id"], "msg": "field required", "type": "missing"}
|
||||
]
|
||||
factory, captured = _make_client(
|
||||
orchestrator_response={"detail": detail_body, "body": None},
|
||||
sdk_response={
|
||||
"verb": "i_am_done",
|
||||
"task_id": "task-A",
|
||||
"attempts": 1,
|
||||
"limit": 3,
|
||||
"window_seconds": 60,
|
||||
"open": False,
|
||||
"circuit_envelope": None,
|
||||
},
|
||||
)
|
||||
with patch("httpx.Client", side_effect=factory):
|
||||
result = flow_module.i_am_done("task-A")
|
||||
assert result["error"] == "incomplete_input"
|
||||
assert isinstance(result["remediate"], str) and result["remediate"]
|
||||
assert result["missing"] == []
|
||||
# The validation detail survives so the agent knows which fields to fix.
|
||||
assert result["detail"] == detail_body
|
||||
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"] == "incomplete_input"
|
||||
|
||||
|
||||
def test_dict_authorized_normalized_to_not_authorized_envelope(
|
||||
flow_module: types.ModuleType,
|
||||
) -> None:
|
||||
"""A dict-shaped PERMISSION_DENIED maps to `error='not_authorized'` in the
|
||||
normalized Envelope (exact-code map, not a substring accident — #161), with
|
||||
the message lifted and a remediate. The breaker counts it as not_authorized."""
|
||||
factory, captured = _make_client(
|
||||
orchestrator_response={
|
||||
"error": {
|
||||
"code": "PERMISSION_DENIED",
|
||||
"message": "you may not merge that PR",
|
||||
"details": {},
|
||||
}
|
||||
},
|
||||
sdk_response={
|
||||
"verb": "i_am_done",
|
||||
"task_id": "task-A",
|
||||
"attempts": 1,
|
||||
"limit": 3,
|
||||
"window_seconds": 60,
|
||||
"open": False,
|
||||
"circuit_envelope": None,
|
||||
},
|
||||
)
|
||||
with patch("httpx.Client", side_effect=factory):
|
||||
result = flow_module.i_am_done("task-A")
|
||||
assert result["error"] == "not_authorized"
|
||||
assert result["message"] == "you may not merge that PR"
|
||||
assert isinstance(result["remediate"], str) and result["remediate"]
|
||||
assert result["missing"] == []
|
||||
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"] == "not_authorized"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# #359: the circuit_open substitution lifts task_id/correlation_id from the
|
||||
# original rejection to the top-level envelope (the SDK's envelope omits them).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_circuit_open_lifts_task_id_and_correlation_id(
|
||||
flow_module: types.ModuleType,
|
||||
) -> None:
|
||||
"""When the breaker trips, the SDK's circuit_envelope carries no task_id /
|
||||
correlation_id, but the original rejection payload does (the orchestrator
|
||||
stamps them on every envelope). The substitution must lift them to the
|
||||
top-level so the agent's envelope contract (read top-level task_id /
|
||||
correlation_id) and ops audit-join still work — not leave them only nested
|
||||
in `inner`."""
|
||||
circuit_env: dict[str, Any] = {
|
||||
"error": "circuit_open",
|
||||
"message": "verb 'i_am_done' rejected 3 times in 60s — breaker open",
|
||||
"remediate": "call i_am_blocked or i_am_idle",
|
||||
}
|
||||
factory, _ = _make_client(
|
||||
orchestrator_response={
|
||||
"error": "tracing_gap",
|
||||
"remediate": "open the PR",
|
||||
"task_id": "T1",
|
||||
"correlation_id": "C1",
|
||||
},
|
||||
sdk_response={
|
||||
"verb": "i_am_done",
|
||||
"task_id": "T1",
|
||||
"attempts": 3,
|
||||
"limit": 3,
|
||||
"window_seconds": 60,
|
||||
"open": True,
|
||||
"circuit_envelope": circuit_env,
|
||||
},
|
||||
)
|
||||
with patch("httpx.Client", side_effect=factory):
|
||||
result = flow_module.i_am_done("task-A")
|
||||
assert result["error"] == "circuit_open"
|
||||
# Lifted from the original rejection to the top level (not just in inner).
|
||||
assert result["task_id"] == "T1"
|
||||
assert result["correlation_id"] == "C1"
|
||||
# The original rejection still survives nested.
|
||||
assert result["inner"]["error"] == "tracing_gap"
|
||||
assert result["inner"]["task_id"] == "T1"
|
||||
|
||||
@@ -70,7 +70,49 @@ async def test_post_draft_reports_http_error() -> None:
|
||||
|
||||
async with _client(handler) as client:
|
||||
result = await intake_server.post_draft("s", {}, client=client)
|
||||
assert result == {"error": "http_503"}
|
||||
# A no-body failure still surfaces the status; ``detail`` is None when the
|
||||
# body is absent/unparseable (#57 — the body is now captured, not dropped).
|
||||
assert result == {"error": "http_503", "detail": None}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_event_captures_body_on_non_success() -> None:
|
||||
"""#57: on a non-success response the relay's body carries the real reason
|
||||
(e.g. 'session not in MegaTask scope' on a 422); _post_event must capture it
|
||||
under ``detail`` so the intake agent gets actionable remediation instead of an
|
||||
opaque ``http_422`` token."""
|
||||
|
||||
def handler(_request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(422, json={"detail": "session not in MegaTask scope"})
|
||||
|
||||
async with _client(handler) as client:
|
||||
result = await intake_server.post_draft("s", {}, client=client)
|
||||
assert result["error"] == "http_422"
|
||||
assert result["detail"] == {"detail": "session not in MegaTask scope"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_propose_batch_result_string_includes_relay_detail(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""The propose_batch failure string surfaces the captured relay body so the
|
||||
grok intake agent can tell whether to re-shape the draft or surface to the
|
||||
CEO (#57). ``post_batch`` is stubbed to the dict _post_event now returns when
|
||||
the relay rejects with a body."""
|
||||
|
||||
async def _relay_rejected(
|
||||
_session_id: str, _payload: dict[str, Any], **_kw: Any
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"error": "http_422",
|
||||
"detail": {"detail": "session not in MegaTask scope"},
|
||||
}
|
||||
|
||||
monkeypatch.setattr(intake_server, "post_batch", _relay_rejected)
|
||||
monkeypatch.setenv("ROBOCO_PROMPTER_SESSION_ID", "sess-1")
|
||||
msg = await intake_server.propose_batch([{"title": "A"}], "MegaTask")
|
||||
assert "Could not submit the MegaTask to the panel" in msg
|
||||
assert "session not in MegaTask scope" in msg
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
||||
Reference in New Issue
Block a user