test(cron): cover verify-on-stop iteration-limit exit normalization

Add turn_finalizer regression tests for unknown/budget_exhausted exits
that must normalize to max_iterations_reached for cron delivery.
This commit is contained in:
HexLab98 2026-07-10 07:10:18 +07:00 committed by kshitij
parent 3eb937a498
commit 53231fb00b

View file

@ -0,0 +1,134 @@
"""Regression tests for iteration-limit exit normalization (#61631)."""
from types import SimpleNamespace
from agent.turn_finalizer import finalize_turn
class _LimitAgent:
def __init__(self, *, max_iterations=60, budget_remaining=0):
self.max_iterations = max_iterations
self.iteration_budget = SimpleNamespace(
remaining=budget_remaining, used=max_iterations, max_total=max_iterations
)
self.quiet_mode = True
self.model = "test-model"
self.provider = "test-provider"
self.base_url = ""
self.session_id = "sess-test"
self.context_compressor = SimpleNamespace(last_prompt_tokens=0)
self.session_input_tokens = 0
self.session_output_tokens = 0
self.session_cache_read_tokens = 0
self.session_cache_write_tokens = 0
self.session_reasoning_tokens = 0
self.session_prompt_tokens = 0
self.session_completion_tokens = 0
self.session_total_tokens = 0
self.session_estimated_cost_usd = 0
self.session_cost_status = "unknown"
self.session_cost_source = "test"
self._tool_guardrail_halt_decision = None
self._interrupt_message = None
self._response_was_previewed = False
self._skill_nudge_interval = 0
self._iters_since_skill = 0
self.valid_tool_names = []
self.persisted_messages = None
self._handle_max_iterations_called = False
def _handle_max_iterations(self, messages, api_call_count):
self._handle_max_iterations_called = True
return "summary from extra call"
def _emit_status(self, *_args, **_kwargs):
pass
def _safe_print(self, *_args, **_kwargs):
pass
def _save_trajectory(self, *_args, **_kwargs):
pass
def _cleanup_task_resources(self, *_args, **_kwargs):
pass
def _drop_trailing_empty_response_scaffolding(self, messages):
pass
def _persist_session(self, messages, conversation_history):
self.persisted_messages = list(messages)
def _file_mutation_verifier_enabled(self):
return False
def _turn_completion_explainer_enabled(self):
return False
def _drain_pending_steer(self):
return None
def clear_interrupt(self):
pass
def _sync_external_memory_for_turn(self, **_kwargs):
pass
def _finalize(agent, *, final_response, exit_reason, api_call_count=60):
return finalize_turn(
agent,
final_response=final_response,
api_call_count=api_call_count,
interrupted=False,
failed=False,
messages=[{"role": "user", "content": "task"}],
conversation_history=[],
effective_task_id="task",
turn_id="turn",
user_message="task",
original_user_message="task",
_should_review_memory=False,
_turn_exit_reason=exit_reason,
)
def test_stale_final_response_unknown_normalizes_for_cron_delivery(monkeypatch):
"""verify-on-stop can leave a composed answer with exit reason ``unknown``."""
monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: [])
agent = _LimitAgent()
report = "complete cron report body"
result = _finalize(agent, final_response=report, exit_reason="unknown")
assert result["final_response"] == report
assert result["turn_exit_reason"] == "max_iterations_reached(60/60)"
assert agent._handle_max_iterations_called is False
def test_stale_final_response_budget_exhausted_normalizes(monkeypatch):
monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: [])
agent = _LimitAgent()
report = "budget exhausted but complete"
result = _finalize(agent, final_response=report, exit_reason="budget_exhausted")
assert result["final_response"] == report
assert result["turn_exit_reason"] == "max_iterations_reached(60/60)"
assert agent._handle_max_iterations_called is False
def test_text_response_exit_not_rewritten_at_iteration_limit(monkeypatch):
monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: [])
agent = _LimitAgent(budget_remaining=5)
exit_reason = "text_response(finish_reason=stop)"
result = _finalize(
agent,
final_response="normal answer",
exit_reason=exit_reason,
api_call_count=59,
)
assert result["turn_exit_reason"] == exit_reason
assert agent._handle_max_iterations_called is False