fix(agent): explain abnormal turn endings instead of blank/partial reply

When a turn ends abnormally after substantive tool calls (empty content
after retries, a partial/truncated stream, exhausted retries, or an
iteration/budget limit), the CLI/TUI response area was left blank or
showed only a fragment (e.g. "The") with no consolidated reason. The
internal turn_exit_reason values (empty_response_exhausted,
partial_stream_recovery, etc.) were never surfaced to the user.

Add a turn-completion explainer that mirrors the existing file-mutation
verifier footer: at turn end, map an abnormal turn_exit_reason to a
short, actionable message and either replace the bare "(empty)"
sentinel or append the reason after a partial fragment. Normal
text_response exits (e.g. a terse "Done.") stay quiet.

Gated by display.turn_completion_explainer (default on) with
HERMES_TURN_COMPLETION_EXPLAINER env override, matching the
file-mutation verifier seam.

Closes #34452
This commit is contained in:
Bartok9
2026-05-29 19:23:05 -07:00
committed by Teknium
parent 897f9533ed
commit 59b0ea98c8
4 changed files with 368 additions and 5 deletions
+18 -5
View File
@@ -3046,7 +3046,11 @@ class TestRunConversation:
mock_compress.assert_not_called() # no compression triggered
assert result["completed"] is True
assert result["final_response"] == "(empty)"
# #34452: the bare "(empty)" sentinel is now replaced by a
# user-visible end-of-turn explanation so the failure isn't silent.
assert result["final_response"] != "(empty)"
assert "without a usable reply" in result["final_response"]
assert result["turn_exit_reason"] == "empty_response_exhausted"
assert result["api_calls"] == 6 # 1 original + 2 prefill + 3 retries
def test_reasoning_only_response_prefill_then_empty(self, agent):
@@ -3066,7 +3070,9 @@ class TestRunConversation:
):
result = agent.run_conversation("answer me")
assert result["completed"] is True
assert result["final_response"] == "(empty)"
# #34452: explanation replaces the bare "(empty)" sentinel.
assert result["final_response"] != "(empty)"
assert "without a usable reply" in result["final_response"]
assert result["api_calls"] == 6 # 1 original + 2 prefill + 3 retries
def test_reasoning_only_prefill_succeeds_on_continuation(self, agent):
@@ -3113,7 +3119,9 @@ class TestRunConversation:
):
result = agent.run_conversation("answer me")
assert result["completed"] is True
assert result["final_response"] == "(empty)"
# #34452: explanation replaces the bare "(empty)" sentinel.
assert result["final_response"] != "(empty)"
assert "without a usable reply" in result["final_response"]
assert result["api_calls"] == 4 # 1 original + 3 retries
def test_truly_empty_response_succeeds_on_nudge(self, agent):
@@ -3209,7 +3217,9 @@ class TestRunConversation:
):
result = agent.run_conversation("answer me")
assert result["completed"] is True
assert result["final_response"] == "(empty)"
# #34452: explanation replaces the bare "(empty)" sentinel.
assert result["final_response"] != "(empty)"
assert "without a usable reply" in result["final_response"]
def test_empty_response_emits_status_for_gateway(self, agent):
"""_emit_status is called during empty retries so gateway users see feedback."""
@@ -3235,7 +3245,10 @@ class TestRunConversation:
):
result = agent.run_conversation("answer me")
assert result["final_response"] == "(empty)"
# #34452: explanation replaces the bare "(empty)" sentinel, but the
# status emissions during retries are unchanged.
assert result["final_response"] != "(empty)"
assert "without a usable reply" in result["final_response"]
# Should have emitted retry statuses (3 retries) + final failure
retry_msgs = [m for m in status_messages if "retrying" in m.lower()]
assert len(retry_msgs) == 3, f"Expected 3 retry status messages, got {len(retry_msgs)}: {status_messages}"
@@ -0,0 +1,181 @@
"""Tests for the end-of-turn completion explainer (#34452).
When a turn ends abnormally after tools (empty content after retries, a
partial/truncated stream, exhausted retries, or an iteration/budget limit)
the user should get a single user-visible explanation of why the reply
stopped instead of a blank or fragmentary response box. Normal short
replies (e.g. ``Done.``) must stay quiet.
These tests exercise:
1. ``_format_turn_completion_explanation`` — the pure reason→message map.
2. ``_turn_completion_explainer_enabled`` — the env/config seam.
3. An end-to-end ``run_conversation`` turn that exhausts empty-response
retries and verifies the explanation reaches ``final_response``.
All assertions work under the mocked OpenAI SDK used elsewhere in this
suite (we patch ``run_agent.OpenAI`` and drive ``agent.client``), so they
pass identically in CI and locally.
"""
import os
import uuid
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
from run_agent import AIAgent
# --------------------------------------------------------------------------
# Fixtures (mirrors tests/run_agent/test_tool_call_guardrail_runtime.py)
# --------------------------------------------------------------------------
def _mock_response(content="Hello", finish_reason="stop", tool_calls=None):
msg = SimpleNamespace(content=content, tool_calls=tool_calls)
choice = SimpleNamespace(message=msg, finish_reason=finish_reason)
return SimpleNamespace(choices=[choice], model="test/model", usage=None)
def _make_agent(max_iterations: int = 10, config: dict | None = None) -> AIAgent:
with (
patch("run_agent.get_tool_definitions", return_value=[]),
patch("run_agent.check_toolset_requirements", return_value={}),
patch("hermes_cli.config.load_config", return_value=config or {}),
patch("run_agent.OpenAI"),
):
agent = AIAgent(
api_key="test-key-1234567890",
base_url="https://openrouter.ai/api/v1",
max_iterations=max_iterations,
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
)
agent.client = MagicMock()
agent._cached_system_prompt = "You are helpful."
agent._use_prompt_caching = False
agent.tool_delay = 0
agent.compression_enabled = False
agent.save_trajectories = False
# No fallback chain so empty responses exhaust deterministically.
agent._fallback_chain = []
return agent
# --------------------------------------------------------------------------
# 1. Pure formatter
# --------------------------------------------------------------------------
def test_explanation_quiet_for_normal_text_response():
"""A healthy text_response exit must NOT produce any explanation."""
out = AIAgent._format_turn_completion_explanation(
"text_response(finish_reason=stop)"
)
assert out == ""
def test_explanation_quiet_for_empty_reason():
assert AIAgent._format_turn_completion_explanation("") == ""
assert AIAgent._format_turn_completion_explanation("unknown") == ""
# guardrail_halt surfaces its own message; explainer stays out of the way.
assert AIAgent._format_turn_completion_explanation("guardrail_halt") == ""
def test_explanation_for_empty_response_exhausted():
out = AIAgent._format_turn_completion_explanation("empty_response_exhausted")
assert out # non-empty
assert "empty content" in out
assert "continue" in out.lower()
def test_explanation_for_partial_stream_recovery():
out = AIAgent._format_turn_completion_explanation("partial_stream_recovery")
assert "partial" in out.lower()
assert "continue" in out.lower()
def test_explanation_for_max_iterations_reached_prefix_match():
"""``max_iterations_reached(...)`` carries a parenthetical suffix."""
out = AIAgent._format_turn_completion_explanation(
"max_iterations_reached(10/10)"
)
assert "iteration" in out.lower()
def test_explanation_for_all_retries_exhausted():
out = AIAgent._format_turn_completion_explanation(
"all_retries_exhausted_no_response"
)
assert "retries" in out.lower()
# --------------------------------------------------------------------------
# 2. Enable/disable seam
# --------------------------------------------------------------------------
def test_explainer_enabled_by_default():
agent = _make_agent()
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("HERMES_TURN_COMPLETION_EXPLAINER", None)
with patch("hermes_cli.config.load_config", return_value={}):
assert agent._turn_completion_explainer_enabled() is True
def test_explainer_disabled_via_env():
agent = _make_agent()
with patch.dict(
os.environ, {"HERMES_TURN_COMPLETION_EXPLAINER": "0"}, clear=False
):
assert agent._turn_completion_explainer_enabled() is False
def test_explainer_disabled_via_config():
agent = _make_agent()
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("HERMES_TURN_COMPLETION_EXPLAINER", None)
with patch(
"hermes_cli.config.load_config",
return_value={"display": {"turn_completion_explainer": False}},
):
assert agent._turn_completion_explainer_enabled() is False
# --------------------------------------------------------------------------
# 3. End-to-end: empty-response exhaustion surfaces the explanation
# --------------------------------------------------------------------------
def test_run_conversation_empty_exhausted_surfaces_explanation():
"""Four empty responses in a row should exhaust retries and the final
response should be the actionable explanation, not a bare '(empty)'."""
agent = _make_agent(max_iterations=10)
# 4 empty responses: retries 1..3 then the terminal on the 4th.
agent.client.chat.completions.create.side_effect = [
_mock_response(content="", finish_reason="stop") for _ in range(8)
]
with (
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("do something")
assert result["turn_exit_reason"] == "empty_response_exhausted"
# The user must NOT be left with a bare sentinel; the explanation wins.
assert result["final_response"] != "(empty)"
assert result["final_response"].strip() != ""
assert "without a usable reply" in result["final_response"]
def test_run_conversation_normal_reply_stays_quiet():
"""A normal short reply like 'Done.' must NOT get an explainer footer."""
agent = _make_agent(max_iterations=10)
agent.client.chat.completions.create.side_effect = [
_mock_response(content="Done.", finish_reason="stop"),
]
with (
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("do something")
assert result["turn_exit_reason"].startswith("text_response")
assert result["final_response"] == "Done."
assert "without a usable reply" not in result["final_response"]