fix(agent): persist recovered final responses

Close a recovery/fallback final_response with an assistant transcript entry before session persistence so durable history cannot end at a tool/user message after the caller receives a final answer.

Adds a regression for a tool-tail transcript with a non-empty final_response. Related to #46071 / #46053, but covers the adjacent case where the assistant message was never appended before persistence.

agent/turn_finalizer.py

@@ -185,6 +185,25 @@ def finalize_turn(
             from agent.message_sanitization import close_interrupted_tool_sequence
             close_interrupted_tool_sequence(messages, final_response)
 
+        # Some recovery/fallback paths return a real final_response without
+        # adding a closing assistant message to the transcript (e.g. the
+        # partial-stream and prior-turn-content recovery ``break`` sites in
+        # ``conversation_loop``). If persisted as-is, the durable session can
+        # end at a tool/user message even though the caller — and the gateway
+        # platform — already saw a completed assistant response. The next turn
+        # then replays a user-only backlog and the model re-answers every
+        # "unanswered" message. Close the durable turn at the source, at the
+        # single chokepoint every recovery ``break`` flows through, so the
+        # invariant "delivered final_response ⇒ assistant row in transcript"
+        # holds regardless of which path produced it. (#43849 / #44100)
+        if final_response and not interrupted:
+            try:
+                _tail_role = messages[-1].get("role") if messages else None
+            except Exception:
+                _tail_role = None
+            if _tail_role != "assistant":
+                messages.append({"role": "assistant", "content": final_response})
+
         agent._persist_session(messages, conversation_history)
     except Exception as _persist_err:
         _cleanup_errors.append(f"persist_session: {_persist_err}")

tests/agent/test_turn_finalizer_final_response_persistence.py

@@ -0,0 +1,112 @@
+from types import SimpleNamespace
+
+from agent.turn_finalizer import finalize_turn
+
+
+class FakeAgent:
+    def __init__(self):
+        self.max_iterations = 90
+        self.iteration_budget = SimpleNamespace(remaining=10, used=1, max_total=90)
+        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 = True
+        self._skill_nudge_interval = 0
+        self._iters_since_skill = 0
+        self.valid_tool_names = []
+        self.persisted_messages = None
+
+    def _handle_max_iterations(self, messages, api_call_count):
+        raise AssertionError("not expected")
+
+    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 test_final_response_closes_tool_tail_before_persistence(monkeypatch):
+    """A recovered/previewed final response must be durable in session history.
+
+    Regression for turns where the caller receives a non-empty final_response,
+    but the message transcript still ends at a tool result. If persisted that
+    way, the next turn reloads a stale/malformed history and can appear to loop
+    because the assistant's visible final answer is missing from durable state.
+    """
+    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: [])
+    agent = FakeAgent()
+    messages = [
+        {"role": "user", "content": "do it"},
+        {
+            "role": "assistant",
+            "content": "I'll check.",
+            "tool_calls": [
+                {"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}
+            ],
+        },
+        {"role": "tool", "tool_call_id": "call-1", "name": "terminal", "content": "ok"},
+    ]
+
+    result = finalize_turn(
+        agent,
+        final_response="Done.",
+        api_call_count=2,
+        interrupted=False,
+        failed=False,
+        messages=messages,
+        conversation_history=[],
+        effective_task_id="task",
+        turn_id="turn",
+        user_message="do it",
+        original_user_message="do it",
+        _should_review_memory=False,
+        _turn_exit_reason="fallback_prior_turn_content",
+    )
+
+    assert result["messages"][-1] == {"role": "assistant", "content": "Done."}
+    assert agent.persisted_messages is not None
+    assert agent.persisted_messages[-1] == {"role": "assistant", "content": "Done."}