Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 62 additions & 2 deletions agent/auxiliary_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -5236,6 +5236,7 @@ def _call_fallback_candidate_sync(
tools=fallback_tools, timeout=effective_timeout,
extra_body=effective_extra_body, reasoning_config=reasoning_config,
base_url=destination.base_url, task=task)
auth_retry_extra_body = effective_extra_body
try:
return _validate_llm_response(
_relay_sync_completion(
Expand All @@ -5247,6 +5248,38 @@ def _call_fallback_candidate_sync(
task,
)
except Exception as fb_err:
# Fallback candidates get the same structured-output degradation as
# the primary path: a provider that rejects ``response_format``
# (DeepSeek: "This response_format type is unavailable now", vLLM
# gateways without xgrammar, strict Anthropic-wire gateways) gets one
# retry without the field. Before this rung the 400 escaped and
# aborted the whole auxiliary task — e.g. title generation timed out
# on the configured local endpoint, fell back to the main agent model
# on DeepSeek, and died on the json_schema field.
if _is_structured_output_rejection(fb_err):
retry_kwargs = _without_structured_output_format(fb_kwargs)
if retry_kwargs is not None:
logger.info(
"Auxiliary %s: fallback candidate %s rejected the "
"structured-output format field; retrying once without it "
"(schema enforcement degrades to prompt compliance): %s",
task or "call", fb_label, fb_err,
)
try:
return _validate_llm_response(
_relay_sync_completion(
fb_client,
retry_kwargs,
provider=destination.provider,
api_mode=destination.api_mode,
),
task,
)
except Exception as retry_err:
if not _is_auth_error(retry_err):
raise
fb_err = retry_err
auth_retry_extra_body = retry_kwargs.get("extra_body")
if not _is_auth_error(fb_err):
raise
fb_provider = _auth_refresh_provider_for_route(
Expand Down Expand Up @@ -5278,7 +5311,7 @@ def _call_fallback_candidate_sync(
retry_messages,
temperature=temperature, max_tokens=max_tokens,
tools=retry_tools, timeout=effective_timeout,
extra_body=effective_extra_body,
extra_body=auth_retry_extra_body,
reasoning_config=reasoning_config,
base_url=retry_destination.base_url, task=task)
try:
Expand Down Expand Up @@ -5342,6 +5375,7 @@ async def _call_fallback_candidate_async(
tools=fallback_tools, timeout=effective_timeout,
extra_body=effective_extra_body, reasoning_config=reasoning_config,
base_url=destination.base_url, task=task)
auth_retry_extra_body = effective_extra_body
try:
return _validate_llm_response(
await _relay_async_completion(
Expand All @@ -5353,6 +5387,32 @@ async def _call_fallback_candidate_async(
task,
)
except Exception as fb_err:
# Structured-output degradation for fallback candidates, mirroring
# _call_fallback_candidate_sync (see its comment for the rationale).
if _is_structured_output_rejection(fb_err):
retry_kwargs = _without_structured_output_format(fb_kwargs)
if retry_kwargs is not None:
logger.info(
"Auxiliary %s: fallback candidate %s rejected the "
"structured-output format field; retrying once without it "
"(schema enforcement degrades to prompt compliance): %s",
task or "call", fb_label, fb_err,
)
try:
return _validate_llm_response(
await _relay_async_completion(
fb_client,
retry_kwargs,
provider=destination.provider,
api_mode=destination.api_mode,
),
task,
)
except Exception as retry_err:
if not _is_auth_error(retry_err):
raise
fb_err = retry_err
auth_retry_extra_body = retry_kwargs.get("extra_body")
if not _is_auth_error(fb_err):
raise
fb_provider = _auth_refresh_provider_for_route(
Expand Down Expand Up @@ -5385,7 +5445,7 @@ async def _call_fallback_candidate_async(
retry_messages,
temperature=temperature, max_tokens=max_tokens,
tools=retry_tools, timeout=effective_timeout,
extra_body=effective_extra_body,
extra_body=auth_retry_extra_body,
reasoning_config=reasoning_config,
base_url=retry_destination.base_url, task=task)
try:
Expand Down
237 changes: 237 additions & 0 deletions tests/agent/test_structured_output_rejection_retry.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@
async_call_llm,
_is_structured_output_rejection,
_without_structured_output_format,
_call_fallback_candidate_sync,
_call_fallback_candidate_async,
)


Expand Down Expand Up @@ -305,3 +307,238 @@ async def test_async_unrelated_400_does_not_retry(self):
},
)
assert client.chat.completions.create.await_count == 1


class _FakeFallbackDestination:
"""Minimal stand-in for ``_FallbackDestination`` used by the fallback
candidate tests: only the attributes the candidate call reads."""

provider = "deepseek"
model = "deepseek-v4-flash"
base_url = "https://api.deepseek.com/v1"
api_mode = "openai"


def _fallback_extra_body():
return {
"response_format": dict(_TITLE_RESPONSE_FORMAT),
"metadata": {"task": "title_generation"},
}


def _fallback_kwargs(
_provider="deepseek",
model="deepseek-v4-flash",
messages=None,
*,
max_tokens=64,
extra_body=None,
**_kwargs,
):
"""Small provider-kwargs builder used by fallback candidate tests."""
return {
"model": model,
"messages": messages or [{"role": "user", "content": "hi"}],
"max_tokens": max_tokens,
"extra_body": dict(
_fallback_extra_body() if extra_body is None else extra_body
),
}


def _status_error(message, status_code):
error = RuntimeError(message)
error.status_code = status_code
return error


def _run_fallback_candidate_sync(relay_side_effect):
"""Drive ``_call_fallback_candidate_sync`` with internals patched out.

Returns ``(result, relay_mock)`` so callers can inspect how many times
the relay was invoked and with which kwargs.
"""
relay = MagicMock(side_effect=relay_side_effect)
with (
patch("agent.auxiliary_client._fallback_entry_timeout",
return_value=None),
patch("agent.auxiliary_client._fallback_destination",
return_value=_FakeFallbackDestination()),
patch("agent.auxiliary_client._replan_synchronous_cache_sections",
side_effect=lambda messages, tools, **kw: (messages, tools)),
patch("agent.auxiliary_client._build_call_kwargs",
side_effect=_fallback_kwargs),
patch("agent.auxiliary_client._relay_sync_completion", relay),
patch("agent.auxiliary_client._validate_llm_response",
side_effect=lambda resp, _task, **_kw: resp),
):
result = _call_fallback_candidate_sync(
MagicMock(), "deepseek-v4-flash", "main-agent(deepseek)",
task="title_generation",
messages=[{"role": "user", "content": "hi"}],
temperature=0.3, max_tokens=64, tools=None,
effective_timeout=30.0,
effective_extra_body=_fallback_extra_body(),
reasoning_config=None,
)
return result, relay


class TestFallbackCandidateStructuredOutputRetry:
"""Fallback candidates get the same structured-output degradation as the
primary path.

Regression: when the configured aux endpoint (e.g. a local Ollama) timed
out, ``call_llm`` fell back to the main agent model. DeepSeek rejects the
``json_schema`` ``response_format`` field with "This response_format type
is unavailable now", and the fallback candidate path re-raised that 400,
killing title generation. It must retry once without the field instead.
"""

def test_retries_once_without_response_format(self):
rejection = RuntimeError(
"Error code: 400 - {'error': {'message': 'This response_format "
"type is unavailable now', 'type': 'invalid_request_error'}}"
)
result, relay = _run_fallback_candidate_sync([rejection, {"ok": True}])

assert result == {"ok": True}
assert relay.call_count == 2
first_kwargs = relay.call_args_list[0].args[1]
retry_kwargs = relay.call_args_list[1].args[1]
assert "response_format" in first_kwargs["extra_body"]
assert "response_format" not in retry_kwargs["extra_body"]
# Sibling extra_body entries survive the scrub.
assert retry_kwargs["extra_body"] == {
"metadata": {"task": "title_generation"},
}

def test_unrelated_400_still_raises(self):
with pytest.raises(RuntimeError, match="Invalid value"):
_run_fallback_candidate_sync(
[RuntimeError("HTTP 400: Invalid value: 'tool'")]
)

def test_rejection_with_retry_also_failing_raises_retry_error(self):
rejection = RuntimeError(
"HTTP 400: This response_format type is unavailable now"
)
retry_error = RuntimeError("Retry failed for a different reason")
with pytest.raises(RuntimeError, match="Retry failed for a different reason"):
_run_fallback_candidate_sync([rejection, retry_error])

def test_auth_refresh_keeps_response_format_removed(self):
refresh_client = MagicMock()
refresh_client.base_url = "https://api.deepseek.com/v1"
with (
patch("agent.auxiliary_client._auth_refresh_provider_for_route",
return_value="deepseek"),
patch("agent.auxiliary_client._refresh_provider_credentials",
return_value=True),
patch("agent.auxiliary_client._get_cached_client",
return_value=(refresh_client, "deepseek-v4-flash")),
):
result, relay = _run_fallback_candidate_sync([
_status_error(
"HTTP 400: This response_format type is unavailable now", 400
),
_status_error("HTTP 401: Unauthorized", 401),
{"ok": True},
])

assert result == {"ok": True}
assert relay.call_count == 3
refreshed_kwargs = relay.call_args_list[2].args[1]
assert "response_format" not in refreshed_kwargs["extra_body"]
assert refreshed_kwargs["extra_body"] == {
"metadata": {"task": "title_generation"},
}


class TestAsyncFallbackCandidateStructuredOutputRetry:
"""Async mirror of the sync fallback-candidate retry semantics."""

async def _run(self, relay_side_effect):
relay = AsyncMock(side_effect=relay_side_effect)
with (
patch("agent.auxiliary_client._fallback_entry_timeout",
return_value=None),
patch("agent.auxiliary_client._fallback_destination",
return_value=_FakeFallbackDestination()),
patch("agent.auxiliary_client._replan_synchronous_cache_sections",
side_effect=lambda messages, tools, **kw: (messages, tools)),
patch("agent.auxiliary_client._build_call_kwargs",
side_effect=_fallback_kwargs),
patch("agent.auxiliary_client._relay_async_completion", relay),
patch("agent.auxiliary_client._validate_llm_response",
side_effect=lambda resp, _task, **_kw: resp),
):
result = await _call_fallback_candidate_async(
MagicMock(), "deepseek-v4-flash", "main-agent(deepseek)",
task="title_generation",
messages=[{"role": "user", "content": "hi"}],
temperature=0.3, max_tokens=64, tools=None,
effective_timeout=30.0,
effective_extra_body=_fallback_extra_body(),
reasoning_config=None,
)
return result, relay

@pytest.mark.asyncio
async def test_async_retries_once_without_response_format(self):
rejection = RuntimeError(
"Error code: 400 - {'error': {'message': 'This response_format "
"type is unavailable now', 'type': 'invalid_request_error'}}"
)
result, relay = await self._run([rejection, {"ok": True}])

assert result == {"ok": True}
assert relay.await_count == 2
first_kwargs = relay.call_args_list[0].args[1]
retry_kwargs = relay.call_args_list[1].args[1]
assert "response_format" in first_kwargs["extra_body"]
assert "response_format" not in retry_kwargs["extra_body"]

@pytest.mark.asyncio
async def test_async_unrelated_400_still_raises(self):
with pytest.raises(RuntimeError, match="Invalid value"):
await self._run(
[RuntimeError("HTTP 400: Invalid value: 'tool'")]
)

@pytest.mark.asyncio
async def test_async_rejection_with_retry_failure_raises_retry_error(self):
rejection = RuntimeError(
"HTTP 400: This response_format type is unavailable now"
)
retry_error = RuntimeError("Retry failed for a different reason")
with pytest.raises(RuntimeError, match="Retry failed for a different reason"):
await self._run([rejection, retry_error])

@pytest.mark.asyncio
async def test_async_auth_refresh_keeps_response_format_removed(self):
refresh_client = MagicMock()
refresh_client.base_url = "https://api.deepseek.com/v1"
with (
patch("agent.auxiliary_client._auth_refresh_provider_for_route",
return_value="deepseek"),
patch("agent.auxiliary_client._refresh_provider_credentials",
return_value=True),
patch("agent.auxiliary_client._get_cached_client",
return_value=(refresh_client, "deepseek-v4-flash")),
):
result, relay = await self._run([
_status_error(
"HTTP 400: This response_format type is unavailable now", 400
),
_status_error("HTTP 401: Unauthorized", 401),
{"ok": True},
])

assert result == {"ok": True}
assert relay.await_count == 3
refreshed_kwargs = relay.call_args_list[2].args[1]
assert "response_format" not in refreshed_kwargs["extra_body"]
assert refreshed_kwargs["extra_body"] == {
"metadata": {"task": "title_generation"},
}