"""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.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_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() # -------------------------------------------------------------------------- # 1b. Cause-aware session-persistence wording # -------------------------------------------------------------------------- def test_explanation_persistence_locked_cause_says_busy_not_disk(): """Write-lock contention must NOT be misdiagnosed as a disk problem.""" out = AIAgent._format_turn_completion_explanation( "session_persistence_failed", "locked" ) lower = out.lower() assert "busy" in lower assert "saved" in lower assert "send it again" in lower assert "disk" not in lower assert "permission" not in lower def test_explanation_persistence_disk_cause_keeps_disk_wording(): out = AIAgent._format_turn_completion_explanation( "session_persistence_failed", "disk" ) lower = out.lower() assert "disk" in lower assert "free some space" in lower or "disk space" in lower def test_explanation_persistence_unknown_cause_is_neutral(): """None/'unknown' cause must not claim disk-full — point at diagnostics.""" for cause in (None, "unknown"): out = AIAgent._format_turn_completion_explanation( "session_persistence_failed", cause ) lower = out.lower() assert out.strip() != "" assert "disk space" not in lower assert "full disk" not in lower assert "hermes doctor" in lower assert "again" in lower def test_explanation_persistence_one_arg_backward_compat(): """Existing one-arg callers must keep working (optional second param).""" out = AIAgent._format_turn_completion_explanation("session_persistence_failed") assert out.strip() != "" assert "session storage" in out.lower() def test_explanation_cause_ignored_for_other_reasons(): """The cause parameter must not perturb non-persistence reasons.""" assert ( AIAgent._format_turn_completion_explanation( "text_response(finish_reason=stop)", "locked" ) == "" ) out = AIAgent._format_turn_completion_explanation( "max_iterations_reached(10/10)", "locked" ) assert "iteration" in out.lower() # -------------------------------------------------------------------------- # 1c. classify_persistence_error — the pure cause classifier # -------------------------------------------------------------------------- def test_classify_persistence_error_categories(): import sqlite3 from hermes_state import classify_persistence_error assert classify_persistence_error( sqlite3.OperationalError("database is locked") ) == "locked" assert classify_persistence_error("SQLITE_BUSY: busy") == "locked" assert classify_persistence_error( sqlite3.OperationalError("database or disk is full") ) == "disk" assert classify_persistence_error("attempt to write a readonly database") == "disk" assert classify_persistence_error("read-only file system") == "disk" assert classify_persistence_error("no space left on device") == "disk" assert classify_persistence_error("disk I/O error") == "disk" assert classify_persistence_error("something else entirely") == "unknown" assert classify_persistence_error(None) == "unknown" assert classify_persistence_error("") == "unknown" def test_classify_persistence_error_reuses_disk_full_markers(): """The disk bucket delegates to hermes_state.is_disk_full_error, so every marker that helper recognizes (ENOSPC, 'not enough space', ...) must classify as 'disk' — the two classifiers can never drift apart.""" import errno from hermes_state import classify_persistence_error assert classify_persistence_error("ENOSPC writing state.db") == "disk" assert classify_persistence_error( "There is not enough space on the disk" ) == "disk" assert classify_persistence_error( OSError(errno.ENOSPC, "No space left on device") ) == "disk" def test_classify_persistence_error_compression_busy_is_locked(): """A live compression lease refusing the write is contention, not storage damage — but its message contains neither 'locked' nor 'busy', so it must classify by exception type (and by phrase for RPC-wrapped strings). This is the exact failure mode of issue #81227.""" from hermes_state import ( CompressionSessionBusyError, SessionCompressionInProgressError, ) from hermes_state import classify_persistence_error assert classify_persistence_error( SessionCompressionInProgressError( "Session 'abc' is being compressed by another writer" ) ) == "locked" assert classify_persistence_error( CompressionSessionBusyError("Compression lease lost before publication: abc") ) == "locked" # RPC-wrapped string forms (exception type lost in transit). assert classify_persistence_error( "Session 'abc' is being compressed by another writer" ) == "locked" assert classify_persistence_error( "Compression lease lost before publication: abc" ) == "locked" def test_persistence_error_causes_tuple_matches_classifier(): """PERSISTENCE_ERROR_CAUSES must cover every value the classifier can return (consumers like cron suppression iterate it).""" from hermes_state import PERSISTENCE_ERROR_CAUSES, classify_persistence_error probes = ( "database is locked", "database or disk is full", "something else entirely", None, ) for probe in probes: assert classify_persistence_error(probe) in PERSISTENCE_ERROR_CAUSES # -------------------------------------------------------------------------- # 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 # -------------------------------------------------------------------------- # 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 "No reply:" in result["final_response"] def test_run_conversation_partial_stream_recovery_surfaces_explanation(): """A long recovered partial stream still needs the visible footer. Without this, the gateway marks the turn as previewed and suppresses the final send, leaving messaging users with a fragment and no reason. """ agent = _make_agent(max_iterations=10) empty_stub = _mock_response(content=None, finish_reason="stop") recovered = ( "I inspected the running gateway and found that the current turn " "stopped after the provider stream timed out." ) def _fake_api_call(_api_kwargs): agent._current_streamed_assistant_text = recovered return empty_stub with ( patch.object(agent, "_interruptible_api_call", side_effect=_fake_api_call), 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"] == "partial_stream_recovery" assert result["final_response"].startswith(recovered) assert "No reply:" in result["final_response"] assert result["response_previewed"] is False