fix(gateway): prevent duplicate final send when only cosmetic edit failed

When the stream consumer's got_done handler successfully delivers the
final response content via _send_or_edit but the subsequent edit
(e.g. cursor removal) fails, final_response_sent remains False even
though the user has already received the final answer. The gateway's
fallback send path then re-delivers the same content, causing the
user to see the response twice on Telegram.

Introduce a new _final_content_delivered flag on the stream consumer,
set by the got_done handler when the final content has reached the
user. The _run_agent suppression logic now treats this flag as an
additional signal (alongside final_response_sent and
response_previewed) that final delivery is already complete.

This preserves the existing behavior for intermediate-text-only
streams (where already_sent=True but no final content has been
delivered) — those still receive the gateway's fallback send, matching
the test expectation in test_partial_stream_output_does_not_set_already_sent.

Adds TestFinalContentDeliveredSuppression with two cases covering
both the suppression (content delivered + edit failed) and the
non-suppression (intermediate text only) branches.

tests/gateway/test_duplicate_reply_suppression.py

@@ -467,3 +467,59 @@ class TestCancellationHandlerDeliveryConfirmation:
             final_response_sent = True
 
         assert final_response_sent is True  # the bug: partial promoted to final
+
+
+class TestFinalContentDeliveredSuppression:
+    """When stream consumer delivered the final content but the cosmetic
+    final edit (cursor removal) failed, the gateway must suppress the
+    fallback send to prevent duplicate messages.
+
+    Covers the scenario not handled by final_response_sent alone:
+    content reached the user via _send_or_edit, but the subsequent edit
+    that clears a typing cursor or streaming marker failed, leaving
+    final_response_sent=False even though the user already saw the text.
+    """
+
+    def test_content_delivered_but_final_edit_failed_suppresses(self):
+        """final_content_delivered=True + final_response_sent=False
+        must suppress (content already visible to user)."""
+        sc = SimpleNamespace(
+            already_sent=True,
+            final_response_sent=False,
+            final_content_delivered=True,
+        )
+        response = {"final_response": "Hello!", "response_previewed": False}
+
+        _streamed = bool(getattr(sc, "final_response_sent", False))
+        _previewed = bool(response.get("response_previewed"))
+        _content_delivered = bool(getattr(sc, "final_content_delivered", False))
+        _is_empty_sentinel = (
+            not response.get("final_response")
+            or response.get("final_response") == "(empty)"
+        )
+        if not _is_empty_sentinel and (_streamed or _previewed or _content_delivered):
+            response["already_sent"] = True
+
+        assert response.get("already_sent") is True
+
+    def test_intermediate_text_only_does_not_suppress(self):
+        """already_sent=True from intermediate text + final_content_delivered=False
+        must NOT suppress (user still needs the real final answer)."""
+        sc = SimpleNamespace(
+            already_sent=True,
+            final_response_sent=False,
+            final_content_delivered=False,
+        )
+        response = {"final_response": "Real answer", "response_previewed": False}
+
+        _streamed = bool(getattr(sc, "final_response_sent", False))
+        _previewed = bool(response.get("response_previewed"))
+        _content_delivered = bool(getattr(sc, "final_content_delivered", False))
+        _is_empty_sentinel = (
+            not response.get("final_response")
+            or response.get("final_response") == "(empty)"
+        )
+        if not _is_empty_sentinel and (_streamed or _previewed or _content_delivered):
+            response["already_sent"] = True
+
+        assert "already_sent" not in response