fix(agent): keep pending verification behind exit provenance

Only restore held verification text when the loop genuinely ends through budget exhaustion. Preserve later interrupts and failures, keep generated-summary fragment explanations, and add regression coverage for both contracts.

agent/turn_finalizer.py

@@ -55,13 +55,20 @@ def finalize_turn(
         api_call_count >= agent.max_iterations
         or agent.iteration_budget.remaining <= 0
     )
+    budget_fallback_eligible = (
+        budget_exhausted
+        and not interrupted
+        and not failed
+        and str(_turn_exit_reason) in {"unknown", "budget_exhausted"}
+    )
     continuation_budget_exhausted = (
         final_response is None
         and bool(_pending_verification_response)
-        and budget_exhausted
+        and budget_fallback_eligible
     )
 
     iteration_limit_fallback = False
+    preserved_verification_fallback = False
     if continuation_budget_exhausted:
         # A verification/continuation gate deliberately withheld a composed
         # answer, then consumed the remaining budget before producing a newer
@@ -71,7 +78,8 @@ def finalize_turn(
         final_response = _pending_verification_response
         _turn_exit_reason = f"max_iterations_reached({api_call_count}/{agent.max_iterations})"
         iteration_limit_fallback = True
-    elif final_response is None and budget_exhausted:
+        preserved_verification_fallback = True
+    elif final_response is None and budget_fallback_eligible:
         # Budget exhausted — ask the model for a summary via one extra
         # API call with tools stripped.  _handle_max_iterations injects a
         # user message and makes a single toolless request.
@@ -320,7 +328,7 @@ def finalize_turn(
                 # truncated partial (the "The" case from #34452).
                 _is_partial_fragment = (
                     not _is_empty_terminal
-                    and not iteration_limit_fallback
+                    and not preserved_verification_fallback
                     and not str(_turn_exit_reason).startswith("text_response")
                     and len(_stripped) <= 24
                     and _stripped[-1:] not in {".", "!", "?", "。", "！", "？", "`", ")"}

tests/agent/test_turn_finalizer_iteration_limit_exit.py

@@ -9,7 +9,13 @@ from agent.turn_finalizer import finalize_turn
 
 
 class _LimitAgent:
-    def __init__(self, *, max_iterations=60, budget_remaining=0):
+    def __init__(
+        self,
+        *,
+        max_iterations=60,
+        budget_remaining=0,
+        completion_explainer=False,
+    ):
         self.max_iterations = max_iterations
         self.iteration_budget = SimpleNamespace(
             remaining=budget_remaining, used=max_iterations, max_total=max_iterations
@@ -39,6 +45,7 @@ class _LimitAgent:
         self.valid_tool_names = []
         self.persisted_messages = None
         self._handle_max_iterations_called = False
+        self._completion_explainer = completion_explainer
 
     def _handle_max_iterations(self, messages, api_call_count):
         self._handle_max_iterations_called = True
@@ -66,7 +73,10 @@ class _LimitAgent:
         return False
 
     def _turn_completion_explainer_enabled(self):
-        return False
+        return self._completion_explainer
+
+    def _format_turn_completion_explanation(self, _reason):
+        return "iteration-limit explanation"
 
     def _drain_pending_steer(self):
         return None
@@ -155,6 +165,30 @@ def test_empty_pending_verification_response_uses_summary_fallback(monkeypatch):
     assert agent._handle_max_iterations_called is True
 
 
+def test_short_generated_summary_keeps_abnormal_turn_explainer(monkeypatch):
+    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: [])
+    agent = _LimitAgent(completion_explainer=True)
+    agent._handle_max_iterations = lambda *_args: "The"
+
+    result = _finalize(agent, final_response=None, exit_reason="unknown")
+
+    assert result["final_response"] == "The\n\niteration-limit explanation"
+
+
+def test_short_preserved_verification_response_is_not_rewritten(monkeypatch):
+    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: [])
+    agent = _LimitAgent(completion_explainer=True)
+
+    result = _finalize(
+        agent,
+        final_response=None,
+        exit_reason="unknown",
+        pending_verification_response="The",
+    )
+
+    assert result["final_response"] == "The"
+
+
 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)
@@ -196,6 +230,43 @@ def test_unrelated_non_success_response_is_not_reclassified(monkeypatch, exit_re
     assert agent._handle_max_iterations_called is False
 
 
+@pytest.mark.parametrize(
+    ("exit_reason", "interrupted", "failed"),
+    [
+        ("interrupted_by_user", True, False),
+        ("all_retries_exhausted_no_response", False, False),
+        ("provider_failure", False, True),
+    ],
+)
+def test_pending_response_does_not_mask_later_terminal_exit(
+    monkeypatch, exit_reason, interrupted, failed
+):
+    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: [])
+    agent = _LimitAgent()
+
+    result = finalize_turn(
+        agent,
+        final_response=None,
+        api_call_count=60,
+        interrupted=interrupted,
+        failed=failed,
+        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,
+        _pending_verification_response="stale premature report",
+    )
+
+    assert result["final_response"] is None
+    assert result["turn_exit_reason"] == exit_reason
+    assert result["completed"] is False
+    assert agent._handle_max_iterations_called is False
+
+
 def test_pending_response_records_kanban_timeout(monkeypatch):
     monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: [])
     monkeypatch.setenv("HERMES_KANBAN_TASK", "task-123")
@@ -213,5 +284,15 @@ def test_pending_response_records_kanban_timeout(monkeypatch):
     )
 
     assert result["turn_exit_reason"] == "max_iterations_reached(60/60)"
-    record.assert_called_once()
-    assert record.call_args.kwargs["outcome"] == "timed_out"
+    record.assert_called_once_with(
+        conn,
+        "task-123",
+        error=(
+            "Iteration budget exhausted (60/60) — task could not complete "
+            "within the allowed iterations"
+        ),
+        outcome="timed_out",
+        release_claim=True,
+        end_run=True,
+        event_payload_extra={"budget_used": 60, "budget_max": 60},
+    )

tests/run_agent/test_verification_continuation_budget.py

@@ -109,7 +109,7 @@ def test_intermediate_ack_uses_summary_instead_of_premature_text(agent, monkeypa
     agent._intent_ack_continuation = True
     agent._looks_like_codex_intermediate_ack = MagicMock(return_value=True)
     agent._interruptible_api_call = lambda _kwargs: _response("I'll inspect the files now")
-    agent._handle_max_iterations = MagicMock(return_value="verified summary")
+    agent._handle_max_iterations = MagicMock(return_value="verified summary.")
     monkeypatch.setenv("HERMES_VERIFY_ON_STOP", "0")
 
     with (
@@ -118,7 +118,7 @@ def test_intermediate_ack_uses_summary_instead_of_premature_text(agent, monkeypa
     ):
         result = agent.run_conversation("inspect /tmp/project")
 
-    assert result["final_response"] == "verified summary"
+    assert result["final_response"] == "verified summary."
     assert result["turn_exit_reason"] == "max_iterations_reached(1/1)"
     agent._handle_max_iterations.assert_called_once()