hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

refactor(turn): sub-split handle_api_error / check_api_response / run_tool_round so no new turn_* function exceeds 300 LOC

Browse Source
This commit is contained in:
Teknium
2026-09-02 17:28:22 -07:00
parent 013c528ef8
commit 7dc43c94d1
3 changed files with 399 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
agent/turn_api_error.py
View File

@@ -80,11 +80,6 @@ def handle_api_error(
"""Recover from ``api_error`` in the original order. Every fallback activation must leave
the retry loop with ``restart_with_rebuilt_messages`` armed (``"break"``) so the pre-API
preflight re-runs against the fallback's context window (#84733)."""
from agent.conversation_loop import (
_arm_fallback_restart,
_is_copilot_provider,
_is_stale_copilot_credential_error,
)
_provider_overflow_recovery_pending = False
def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ApiErrorVerdict:
@@ -313,6 +308,96 @@ def handle_api_error(
if _ov.action == "continue":
return _verdict("continue")
_ue = settle_unrecovered_error(
agent,
api_error=api_error,
classified=classified,
_retry=_retry,
status_code=status_code,
error_msg=error_msg,
is_context_length_error=is_context_length_error,
is_rate_limited=is_rate_limited,
_is_zai_coding_overload=_is_zai_coding_overload,
_provider=_provider,
_base=_base,
_model=_model,
messages=messages,
api_messages=api_messages,
api_kwargs=api_kwargs,
active_system_prompt=active_system_prompt,
conversation_history=conversation_history,
approx_tokens=approx_tokens,
retry_count=retry_count,
max_retries=max_retries,
compression_attempts=compression_attempts,
api_call_count=api_call_count,
)
active_system_prompt = _ue.active_system_prompt
retry_count = _ue.retry_count
compression_attempts = _ue.compression_attempts
if _ue.action != "fallthrough":
return _verdict(_ue.action, _ue.result)
return _verdict("fallthrough")
@dataclass
class UnrecoveredErrorVerdict:
"""``action``: ``"continue"`` (retry), ``"break"`` (fallback armed / redirect pending) or
``"return"`` (``result`` is the terminal result dict). Rebinds ``active_system_prompt``,
``retry_count`` and ``compression_attempts``."""
action: str
active_system_prompt: Any
retry_count: Any
compression_attempts: Any
result: Optional[Dict[str, Any]] = None
def settle_unrecovered_error(
agent: Any,
*,
api_error: Any,
classified: Any,
_retry: Any,
status_code: Any,
error_msg: Any,
is_context_length_error: Any,
is_rate_limited: Any,
_is_zai_coding_overload: Any,
_provider: Any,
_base: Any,
_model: Any,
messages: Any,
api_messages: Any,
api_kwargs: Any,
active_system_prompt: Any,
conversation_history: Any,
approx_tokens: Any,
retry_count: Any,
max_retries: Any,
compression_attempts: Any,
api_call_count: Any,
) -> UnrecoveredErrorVerdict:
"""Decide the fate of an API error that every recovery chain declined: local validation /
non-retryable client errors (Copilot stale-credential self-heal first, then fallback, then a
terminal result), max-retries exhaustion (primary transport recovery -> fallback -> terminal
result), else the interruptible error backoff. ``FailoverReason.billing`` (402) is deliberately
treated as non-retryable (#31273)."""
from agent.conversation_loop import (
_arm_fallback_restart,
_is_copilot_provider,
_is_stale_copilot_credential_error,
)
def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> UnrecoveredErrorVerdict:
return UnrecoveredErrorVerdict(
action=action,
active_system_prompt=active_system_prompt,
retry_count=retry_count,
compression_attempts=compression_attempts,
result=result,
)
# Non-retryable: ValueError/TypeError are local bugs, except
# UnicodeEncodeError (surrogate path above) and json.JSONDecodeError, a
# transient provider/network failure that must be retried (#14782).

271
agent/turn_response_check.py
View File

@@ -74,10 +74,6 @@ def check_api_response(
(bytes back != usable content); ``_preflight_compression_blocked``/``_last_preflight_pressure``
reset only when the usage fold re-arms the compression budget."""
from agent.conversation_loop import (
_arm_fallback_restart,
describe_invalid_response,
jittered_backoff,
interruptible_backoff_sleep,
validate_response_shape,
)
api_duration = None
@@ -122,101 +118,33 @@ def check_api_response(
response_invalid, error_details = validate_response_shape(agent, response)
if response_invalid:
agent._invoke_api_request_error_hook(
task_id=effective_task_id,
turn_id=turn_id,
api_request_id=api_request_id,
api_call_count=api_call_count,
api_start_time=api_start_time,
_iv = retry_invalid_response(
agent,
response=response,
error_details=error_details,
_retry=_retry,
thinking_spinner=thinking_spinner,
messages=messages,
api_messages=api_messages,
api_kwargs=api_kwargs,
error_type="InvalidAPIResponse",
error_message=", ".join(error_details) or "Invalid API response",
status_code=getattr(getattr(response, "error", None), "code", None),
active_system_prompt=active_system_prompt,
conversation_history=conversation_history,
retry_count=retry_count,
max_retries=max_retries,
retryable=True,
reason="invalid_response",
)
# Stop spinner silently — retry status is now buffered
# and only surfaced if every retry+fallback exhausts.
if thinking_spinner:
thinking_spinner.stop("")
thinking_spinner = None
if agent.thinking_callback:
agent.thinking_callback("")
# Invalid response — could be rate limiting, provider timeout,
# upstream server error, or malformed response.
retry_count += 1
# Eager fallback: empty/malformed responses often mean rate limiting
# — switch now instead of extended backoff.
if agent._fallback_index < len(agent._fallback_chain):
agent._buffer_status("⚠️ Empty/malformed response — switching to fallback...")
if agent._try_activate_fallback():
active_system_prompt = _arm_fallback_restart(
agent, api_messages, active_system_prompt, _retry)
retry_count = 0
compression_attempts = 0
return _verdict("break")
error_msg, provider_name, _failure_hint = describe_invalid_response(
agent, response, api_duration
)
agent._buffer_vprint(f"⚠️ Invalid API response (attempt {retry_count}/{max_retries}): {', '.join(error_details)}")
agent._buffer_vprint(f" 🏢 Provider: {provider_name}")
cleaned_provider_error = agent._clean_error_message(error_msg)
agent._buffer_vprint(f" 📝 Provider message: {cleaned_provider_error}")
agent._buffer_vprint(f" ⏱️ {_failure_hint}")
if retry_count >= max_retries:
# Try fallback before giving up
if agent._has_pending_fallback():
agent._buffer_status(f"⚠️ Max retries ({max_retries}) for invalid responses — trying fallback...")
if agent._try_activate_fallback():
active_system_prompt = _arm_fallback_restart(
agent, api_messages, active_system_prompt, _retry)
retry_count = 0
compression_attempts = 0
return _verdict("break")
# Terminal — flush buffered retry trace so user sees what happened.
agent._flush_status_buffer()
agent._emit_status(f"❌ Max retries ({max_retries}) exceeded for invalid responses. Giving up.")
logger.error("%sInvalid API response after %d retries.", agent.log_prefix, max_retries)
agent._persist_session(messages, conversation_history)
_final_response = f"Invalid API response after {max_retries} retries: {_failure_hint}"
return _verdict("return", {
"final_response": _final_response,
"messages": messages,
"completed": False,
"api_calls": api_call_count,
"error": _final_response,
"failed": True # Mark as failure for filtering
})
# Backoff before retry — jittered exponential: 5s base, 120s cap
wait_time = jittered_backoff(retry_count, base_delay=5.0, max_delay=120.0)
agent._buffer_vprint(f"⏳ Retrying in {wait_time:.1f}s ({_failure_hint})...")
logger.warning("Invalid API response (retry %d/%d): %s | Provider: %s", retry_count, max_retries, ', '.join(error_details), provider_name)
# A redirect cancels only the live request; the helper preserves the
# pending correction (restart_with_redirected_messages) instead of
# destroying it with clear_interrupt().
_interrupted = interruptible_backoff_sleep(
agent, wait_time, _retry,
messages=messages,
conversation_history=conversation_history,
compression_attempts=compression_attempts,
api_call_count=api_call_count,
abort_message="Interrupt detected during retry wait, aborting.",
interrupt_text=f"Operation interrupted during retry ({_failure_hint}, attempt {retry_count}/{max_retries}).",
activity_label=f"retry backoff ({retry_count}/{max_retries})",
api_request_id=api_request_id,
api_start_time=api_start_time,
api_duration=api_duration,
effective_task_id=effective_task_id,
turn_id=turn_id,
)
if _interrupted is not None:
return _verdict("return", _interrupted)
if _retry.restart_with_redirected_messages:
return _verdict("break") # rebuild this iteration from the correction
return _verdict("continue") # Retry the API call
thinking_spinner = _iv.thinking_spinner
active_system_prompt = _iv.active_system_prompt
retry_count = _iv.retry_count
compression_attempts = _iv.compression_attempts
if _iv.action != "fallthrough":
return _verdict(_iv.action, _iv.result)
agent._turn_received_provider_response = True
@@ -362,3 +290,158 @@ def check_api_response(
agent._touch_activity(f"API call #{api_call_count} completed")
return _verdict("break") # Success, exit retry loop
return _verdict("fallthrough")
@dataclass
class InvalidResponseVerdict:
"""``action``: ``"continue"`` (retry the API call after backoff), ``"break"`` (fallback
armed / redirect pending) or ``"return"`` (``result``: terminal invalid-response result or
interrupt during backoff). Rebinds ``thinking_spinner``/``active_system_prompt``/
``retry_count``/``compression_attempts``."""
action: str
thinking_spinner: Any
active_system_prompt: Any
retry_count: Any
compression_attempts: Any
result: Optional[Dict[str, Any]] = None
def retry_invalid_response(
agent: Any,
*,
response: Any,
error_details: Any,
_retry: Any,
thinking_spinner: Any,
messages: Any,
api_messages: Any,
api_kwargs: Any,
active_system_prompt: Any,
conversation_history: Any,
retry_count: Any,
max_retries: Any,
compression_attempts: Any,
api_call_count: Any,
api_request_id: Any,
api_start_time: Any,
api_duration: Any,
effective_task_id: Any,
turn_id: Any,
) -> InvalidResponseVerdict:
"""Malformed/empty provider response: fire the error hook, stop the spinner, eager
fallback (empty responses often mean rate limiting), terminal result at max retries,
else jittered backoff that preserves a pending redirect."""
from agent.conversation_loop import (
_arm_fallback_restart,
describe_invalid_response,
interruptible_backoff_sleep,
jittered_backoff,
)
def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> InvalidResponseVerdict:
return InvalidResponseVerdict(
action=action,
thinking_spinner=thinking_spinner,
active_system_prompt=active_system_prompt,
retry_count=retry_count,
compression_attempts=compression_attempts,
result=result,
)
agent._invoke_api_request_error_hook(
task_id=effective_task_id,
turn_id=turn_id,
api_request_id=api_request_id,
api_call_count=api_call_count,
api_start_time=api_start_time,
api_kwargs=api_kwargs,
error_type="InvalidAPIResponse",
error_message=", ".join(error_details) or "Invalid API response",
status_code=getattr(getattr(response, "error", None), "code", None),
retry_count=retry_count,
max_retries=max_retries,
retryable=True,
reason="invalid_response",
)
# Stop spinner silently — retry status is now buffered
# and only surfaced if every retry+fallback exhausts.
if thinking_spinner:
thinking_spinner.stop("")
thinking_spinner = None
if agent.thinking_callback:
agent.thinking_callback("")
# Invalid response — could be rate limiting, provider timeout,
# upstream server error, or malformed response.
retry_count += 1
# Eager fallback: empty/malformed responses often mean rate limiting
# — switch now instead of extended backoff.
if agent._fallback_index < len(agent._fallback_chain):
agent._buffer_status("⚠️ Empty/malformed response — switching to fallback...")
if agent._try_activate_fallback():
active_system_prompt = _arm_fallback_restart(
agent, api_messages, active_system_prompt, _retry)
retry_count = 0
compression_attempts = 0
return _verdict("break")
error_msg, provider_name, _failure_hint = describe_invalid_response(
agent, response, api_duration
)
agent._buffer_vprint(f"⚠️ Invalid API response (attempt {retry_count}/{max_retries}): {', '.join(error_details)}")
agent._buffer_vprint(f" 🏢 Provider: {provider_name}")
cleaned_provider_error = agent._clean_error_message(error_msg)
agent._buffer_vprint(f" 📝 Provider message: {cleaned_provider_error}")
agent._buffer_vprint(f" ⏱️ {_failure_hint}")
if retry_count >= max_retries:
# Try fallback before giving up
if agent._has_pending_fallback():
agent._buffer_status(f"⚠️ Max retries ({max_retries}) for invalid responses — trying fallback...")
if agent._try_activate_fallback():
active_system_prompt = _arm_fallback_restart(
agent, api_messages, active_system_prompt, _retry)
retry_count = 0
compression_attempts = 0
return _verdict("break")
# Terminal — flush buffered retry trace so user sees what happened.
agent._flush_status_buffer()
agent._emit_status(f"❌ Max retries ({max_retries}) exceeded for invalid responses. Giving up.")
logger.error("%sInvalid API response after %d retries.", agent.log_prefix, max_retries)
agent._persist_session(messages, conversation_history)
_final_response = f"Invalid API response after {max_retries} retries: {_failure_hint}"
return _verdict("return", {
"final_response": _final_response,
"messages": messages,
"completed": False,
"api_calls": api_call_count,
"error": _final_response,
"failed": True # Mark as failure for filtering
})
# Backoff before retry — jittered exponential: 5s base, 120s cap
wait_time = jittered_backoff(retry_count, base_delay=5.0, max_delay=120.0)
agent._buffer_vprint(f"⏳ Retrying in {wait_time:.1f}s ({_failure_hint})...")
logger.warning("Invalid API response (retry %d/%d): %s | Provider: %s", retry_count, max_retries, ', '.join(error_details), provider_name)
# A redirect cancels only the live request; the helper preserves the
# pending correction (restart_with_redirected_messages) instead of
# destroying it with clear_interrupt().
_interrupted = interruptible_backoff_sleep(
agent, wait_time, _retry,
messages=messages,
conversation_history=conversation_history,
api_call_count=api_call_count,
abort_message="Interrupt detected during retry wait, aborting.",
interrupt_text=f"Operation interrupted during retry ({_failure_hint}, attempt {retry_count}/{max_retries}).",
activity_label=f"retry backoff ({retry_count}/{max_retries})",
)
if _interrupted is not None:
return _verdict("return", _interrupted)
if _retry.restart_with_redirected_messages:
return _verdict("break") # rebuild this iteration from the correction
return _verdict("continue") # Retry the API call
return _verdict("fallthrough")

220
agent/turn_tool_round.py
View File

@@ -62,7 +62,6 @@ def run_tool_round(
Hermes; a failed canonical append ends the turn rather than running tools from
process-only state."""
from agent.conversation_loop import (
_STALE_MARKER_RE,
_invalid_tool_name_error_content,
)
@@ -122,94 +121,14 @@ def run_tool_round(
if tc.function.name not in agent.valid_tool_names
]
assistant_msg = agent._build_assistant_message(assistant_message, finish_reason)
turn_content = assistant_message.content or ""
# A bare bracketed token (e.g. ``[memory]``) beside a function call is
# protocol scaffolding; persisting it lets the post-tool fallback replay
# it forever (#78148).
if (
assistant_message.tool_calls
and _STALE_MARKER_RE.fullmatch(turn_content.strip())
):
logger.warning(
"Discarding bare tool-call marker from assistant content: %s",
turn_content,
)
turn_content = ""
assistant_msg["content"] = ""
# Classify tools regardless of visible content: a substantive tool-only
# turn must invalidate any older housekeeping fallback.
_HOUSEKEEPING_TOOLS = frozenset({
"memory", "todo_list", "skill_manage", "session_search",
})
_all_housekeeping = all(
tc.function.name in _HOUSEKEEPING_TOOLS
for tc in assistant_message.tool_calls
)
# Substantive tools clear any older fallback so a two-turn-old
# housekeeping narration isn't attributed to the preceding tool turn.
if assistant_message.tool_calls and not _all_housekeeping:
agent._last_content_with_tools = None
agent._last_content_tools_all_housekeeping = False
# Also clear the mute flag a prior housekeeping turn may have set,
# else _vprint suppresses this turn's tool progress until the
# no-tool-call branch clears it.
agent._mute_post_response = False
# Content + tool_calls in one turn: keep the content as a fallback final
# response in case the follow-up turn after tools is empty.
if turn_content and agent._has_content_after_think_block(turn_content):
agent._last_content_with_tools = turn_content
# Mute only when EVERY tool call is post-response housekeeping
# (memory, todo, skill_manage); substantive tools keep output on.
agent._last_content_tools_all_housekeeping = _all_housekeeping
if _all_housekeeping and agent._has_stream_consumers():
agent._mute_post_response = True
elif agent._should_emit_quiet_tool_messages():
clean = agent._strip_think_blocks(turn_content).strip()
if clean:
agent._vprint(f" ┊ 💬 {clean}")
# Pop thinking-only prefill message(s) before appending
# (tool-call path — same rationale as the final-response path).
_had_prefill = False
while (
messages
and isinstance(messages[-1], dict)
and messages[-1].get("_thinking_prefill")
):
messages.pop()
_had_prefill = True
# Tool calls after a prefill recovery reset the prefill counter, so
# each tool-call success is a fresh start, not a cumulative burn.
if _had_prefill:
agent._thinking_prefill_retries = 0
agent._empty_content_retries = 0
# Re-arm the post-tool nudge so it can fire on a LATER tool round.
agent._post_tool_empty_retried = False
# A landed tool call recovers any dropped-tool-call stall; refresh that
# budget so it guards each stall independently, not the whole run.
agent._dropped_toolcall_retries = 0
previous_msg = messages[-1] if messages else None
current_interim_visible = agent._interim_assistant_visible_text(assistant_msg)
previous_interim_visible = (
agent._interim_assistant_visible_text(previous_msg)
if isinstance(previous_msg, dict)
else ""
)
duplicate_previous_interim = (
bool(current_interim_visible)
and isinstance(previous_msg, dict)
and previous_msg.get("role") == "assistant"
and previous_msg.get("finish_reason") == "incomplete"
and previous_interim_visible == current_interim_visible
_st = stage_tool_call_message(
agent,
assistant_message=assistant_message,
finish_reason=finish_reason,
messages=messages,
)
assistant_msg = _st.assistant_msg
duplicate_previous_interim = _st.duplicate_previous_interim
append_message(messages, assistant_msg)
# Mixed batch: error-result invalid calls and drop them from execution.
@@ -348,3 +267,128 @@ def run_tool_round(
# Continue loop for next response
return _verdict("continue")
return _verdict("fallthrough")
@dataclass
class StagedToolCallMessage:
"""Always ``action == "fallthrough"``. ``assistant_msg`` is the transcript row to append;
``duplicate_previous_interim`` suppresses re-emitting interim commentary the previous
``incomplete`` row already showed."""
action: str
assistant_msg: Any
duplicate_previous_interim: Any
def stage_tool_call_message(
agent: Any,
*,
assistant_message: Any,
finish_reason: Any,
messages: Any,
) -> StagedToolCallMessage:
"""Build the assistant tool-call row and update the per-turn fallback/mute state: drop a bare
bracketed marker beside a call (#78148), classify housekeeping-only rounds, keep visible
content as the empty-follow-up fallback, pop thinking-only prefills (resetting their
counters), re-arm the post-tool nudge and the dropped-tool-call stall budget."""
from agent.conversation_loop import (
_STALE_MARKER_RE,
)
def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> StagedToolCallMessage:
return StagedToolCallMessage(
action=action,
assistant_msg=assistant_msg,
duplicate_previous_interim=duplicate_previous_interim,
)
assistant_msg = agent._build_assistant_message(assistant_message, finish_reason)
turn_content = assistant_message.content or ""
# A bare bracketed token (e.g. ``[memory]``) beside a function call is
# protocol scaffolding; persisting it lets the post-tool fallback replay
# it forever (#78148).
if (
assistant_message.tool_calls
and _STALE_MARKER_RE.fullmatch(turn_content.strip())
):
logger.warning(
"Discarding bare tool-call marker from assistant content: %s",
turn_content,
)
turn_content = ""
assistant_msg["content"] = ""
# Classify tools regardless of visible content: a substantive tool-only
# turn must invalidate any older housekeeping fallback.
_HOUSEKEEPING_TOOLS = frozenset({
"memory", "todo_list", "skill_manage", "session_search",
})
_all_housekeeping = all(
tc.function.name in _HOUSEKEEPING_TOOLS
for tc in assistant_message.tool_calls
)
# Substantive tools clear any older fallback so a two-turn-old
# housekeeping narration isn't attributed to the preceding tool turn.
if assistant_message.tool_calls and not _all_housekeeping:
agent._last_content_with_tools = None
agent._last_content_tools_all_housekeeping = False
# Also clear the mute flag a prior housekeeping turn may have set,
# else _vprint suppresses this turn's tool progress until the
# no-tool-call branch clears it.
agent._mute_post_response = False
# Content + tool_calls in one turn: keep the content as a fallback final
# response in case the follow-up turn after tools is empty.
if turn_content and agent._has_content_after_think_block(turn_content):
agent._last_content_with_tools = turn_content
# Mute only when EVERY tool call is post-response housekeeping
# (memory, todo, skill_manage); substantive tools keep output on.
agent._last_content_tools_all_housekeeping = _all_housekeeping
if _all_housekeeping and agent._has_stream_consumers():
agent._mute_post_response = True
elif agent._should_emit_quiet_tool_messages():
clean = agent._strip_think_blocks(turn_content).strip()
if clean:
agent._vprint(f" ┊ 💬 {clean}")
# Pop thinking-only prefill message(s) before appending
# (tool-call path — same rationale as the final-response path).
_had_prefill = False
while (
messages
and isinstance(messages[-1], dict)
and messages[-1].get("_thinking_prefill")
):
messages.pop()
_had_prefill = True
# Tool calls after a prefill recovery reset the prefill counter, so
# each tool-call success is a fresh start, not a cumulative burn.
if _had_prefill:
agent._thinking_prefill_retries = 0
agent._empty_content_retries = 0
# Re-arm the post-tool nudge so it can fire on a LATER tool round.
agent._post_tool_empty_retried = False
# A landed tool call recovers any dropped-tool-call stall; refresh that
# budget so it guards each stall independently, not the whole run.
agent._dropped_toolcall_retries = 0
previous_msg = messages[-1] if messages else None
current_interim_visible = agent._interim_assistant_visible_text(assistant_msg)
previous_interim_visible = (
agent._interim_assistant_visible_text(previous_msg)
if isinstance(previous_msg, dict)
else ""
)
duplicate_previous_interim = (
bool(current_interim_visible)
and isinstance(previous_msg, dict)
and previous_msg.get("role") == "assistant"
and previous_msg.get("finish_reason") == "incomplete"
and previous_interim_visible == current_interim_visible
)
return _verdict("fallthrough")