fix(stream): tidy clean-EOF diagnostics and truncation copy

- _mark_finish_seen helper on the local _diag; also set on the Anthropic
  path when message_delta carries stop_reason
- emit_stream_drop status: 'attempt N/M dropped, reconnecting'
- clean-EOF log wording: server or proxy closed the stream cleanly
- truncated_unreported copy drops the raw finish_reason placeholder
- turn_tool_validation compares against FINISH_REASON_LENGTH

agent/chat_completion_helpers.py

@@ -2982,6 +2982,12 @@ class _StreamingCall(StreamingWaitMonitor):
             # Delta-length estimate: ~3x cheaper than repr() per chunk.
             diag["bytes"] = int(diag.get("bytes", 0)) + _estimate_chunk_bytes(chunk)
 
+    @staticmethod
+    def _mark_finish_seen(diag, finish_reason) -> None:
+        """Record that this attempt saw a terminal finish/stop reason (#102766)."""
+        if finish_reason and isinstance(diag, dict) and not diag.get("finish_reason_seen"):
+            diag["finish_reason_seen"] = True
+
     # ── chat_completions wire ───────────────────────────────────────────
 
     def _stream_timeouts(self) -> tuple[float, float, float]:
@@ -3176,8 +3182,7 @@ class _StreamingCall(StreamingWaitMonitor):
             if not chunk.choices:
                 usage, finish_reason = self._choiceless_chunk(chunk, finish_reason)
                 usage_obj = usage or usage_obj
-                if finish_reason and isinstance(self.clients.diag, dict):
-                    self.clients.diag["finish_reason_seen"] = True  # #102766
+                self._mark_finish_seen(_diag, finish_reason)
                 continue
 
             choice = chunk.choices[0]
@@ -3185,8 +3190,7 @@ class _StreamingCall(StreamingWaitMonitor):
             # Read finish_reason/usage BEFORE any content-shape `continue`: the SSE-echo
             # guard can swallow a merged finish chunk (vLLM standalone ':' tokens).
             finish_reason = _normalize_finish_reason(getattr(choice, "finish_reason", None)) or finish_reason
-            if finish_reason and isinstance(self.clients.diag, dict):
-                self.clients.diag["finish_reason_seen"] = True  # #102766
+            self._mark_finish_seen(_diag, finish_reason)
             if hasattr(chunk, "usage") and chunk.usage:
                 usage_obj = chunk.usage
 
@@ -3345,8 +3349,8 @@ class _StreamingCall(StreamingWaitMonitor):
             _dropped_names = [(tool_calls_acc[idx]["function"]["name"] or "?") for idx in sorted(tool_calls_acc)]
             logger.warning(
                 "Clean EOF, no finish_reason: server ended the stream (no transport exception) while a tool "
-                "call's arguments were still incomplete (tools=%s). Not a network drop and not an "
-                "output-length truncation.",
+                "call's arguments were still incomplete (tools=%s). The server or a proxy closed the stream "
+                "cleanly; not an output-length truncation.",
                 _dropped_names)
             return _build_partial_stream_stub(
                 role, full_content, full_reasoning, model_name, usage_obj, dropped_tool_names=_dropped_names or None,
@@ -3358,7 +3362,7 @@ class _StreamingCall(StreamingWaitMonitor):
             # A usage object proves the provider finished (include_usage's final chunk).
             logger.warning(
                 "Clean EOF, no finish_reason: server ended the stream (no transport exception) after delivering "
-                "text with no tool calls. Not a network drop.")
+                "text with no tool calls. The server or a proxy closed the stream cleanly.")
             return _build_partial_stream_stub(role, full_content, full_reasoning, model_name, usage_obj, clean_eof=True)
         effective_finish_reason = "length" if has_truncated_tool_args else (finish_reason or "stop")
         provider_stream_error = _provider_stream_error_from_text(
@@ -3454,7 +3458,9 @@ class _StreamingCall(StreamingWaitMonitor):
                 event_type = getattr(event, "type", None)
                 if event_type == "message_stop":
                     saw_message_stop = True
-                if event_type == "content_block_start":
+                elif event_type == "message_delta":
+                    self._mark_finish_seen(_diag, getattr(getattr(event, "delta", None), "stop_reason", None))
+                elif event_type == "content_block_start":
                     block = getattr(event, "content_block", None)
                     if block and getattr(block, "type", None) == "tool_use":
                         has_tool_use = True

agent/stream_diag.py

@@ -164,7 +164,7 @@ def emit_stream_drop(
     try:
         agent._buffer_diagnostic_status(
             f"⚠️ {provider} stream {kind} ({type(error).__name__}){_suffix} "
-            f"— reconnecting, retry {attempt}/{max_attempts}"
+            f"— attempt {attempt}/{max_attempts} dropped, reconnecting"
         )
         agent._touch_activity(f"stream retry {attempt}/{max_attempts} after {type(error).__name__}")
     except Exception:

agent/turn_tool_validation.py

@@ -17,6 +17,7 @@ from typing import Any, Dict, List, Optional
 from agent.message_metadata import append_message
 from agent.message_sanitization import close_interrupted_tool_sequence, coalesce_tool_call_id
 from agent.turn_failure_copy import site_copy, stamp_failure
+from hermes_constants import FINISH_REASON_LENGTH
 
 logger = logging.getLogger("agent.conversation_loop")
 
@@ -180,8 +181,8 @@ def validate_tool_calls(
             # Blame the output cap only when the model reported one; otherwise the args
             # were cut by a stream break or a router rewriting finish_reason (#91717).
             _copy = (
-                site_copy("truncated") if finish_reason == "length"
-                else site_copy("truncated_unreported", finish_reason=repr(finish_reason))
+                site_copy("truncated") if finish_reason == FINISH_REASON_LENGTH
+                else site_copy("truncated_unreported")
             )
             return _verdict("return", _partial_exit(
                 agent, messages, conversation_history, api_call_count, _copy,

tests/agent/test_partial_stream_finish_reason.py

@@ -1162,8 +1162,8 @@ class TestRouterRewriteTruncationMessageIsHonest:
             "output-length limit (#91717)."
         )
         assert "output length limit" not in error
-        # The honest message names the real suspect and surfaces finish_reason.
-        assert "finish_reason='tool_calls'" in final
+        # The honest message names the real suspect (raw finish_reason stays in the diagnostic log).
+        assert "finish_reason" not in final
         assert (
             "transport" in final or "router" in final
         ), "Honest message must point at transport/router corruption."

tests/agent/test_stream_drop_logging.py

@@ -124,6 +124,7 @@ def test_retry_after_drop_reports_the_attempt_that_dropped():
         agent=agent,
         clients=SimpleNamespace(diag=None, close_once=lambda reason: None),
         _cancel_current_stream_attempt=lambda reason: None,
+        last_chunk_time={"t": 0.0},
     )
     _StreamingCall._retry_after_drop(fake, ConnectionError("drop"), 0, 2, mid_tool_call=False, reason="t")
     assert agent._emit_stream_drop.call_args.kwargs["attempt"] == 1