fix: propagate response_transformed flag — plugin hook output survives streaming suppression

When a transform_llm_output hook modifies final_response after streaming,
the gateway was silently discarding the transformed content because
streamed=True / content_delivered=True triggered the final-send
suppression. Three changes:

1. conversation_loop: set `_response_transformed=True` when a
   transform_llm_output hook returns a non-empty string, and expose it
   as `response_transformed` in the result dict.

2. gateway/run: skip the final-send suppression when
   `response_transformed` is True — the transformed response must
   reach the client even if streaming already sent the original text.

3. acp_adapter/server: remove `not streamed_message` guard so
   final_response is always delivered (ACP path fixed separately).

agent/conversation_loop.py

@@ -4031,6 +4031,8 @@ def run_conversation(
         except Exception as _ver_err:
             logger.debug("file-mutation verifier footer failed: %s", _ver_err)
 
+    _response_transformed = False
+
     # Plugin hook: transform_llm_output
     # Fired once per turn after the tool-calling loop completes.
     # Plugins can transform the LLM's output text before it's returned.
@@ -4045,9 +4047,11 @@ def run_conversation(
                 model=agent.model,
                 platform=getattr(agent, "platform", None) or "",
             )
+            _response_transformed = False
             for _hook_result in _transform_results:
                 if isinstance(_hook_result, str) and _hook_result:
                     final_response = _hook_result
+                    _response_transformed = True
                     break  # First non-empty string wins
         except Exception as exc:
             logger.warning("transform_llm_output hook failed: %s", exc)
@@ -4099,6 +4103,7 @@ def run_conversation(
         "failed": failed,
         "partial": False,  # True only when stopped due to invalid tool calls
         "interrupted": interrupted,
+        "response_transformed": _response_transformed,
         "response_previewed": getattr(agent, "_response_was_previewed", False),
         "model": agent.model,
         "provider": agent.provider,

gateway/run.py

@@ -17678,7 +17678,11 @@ class GatewayRunner:
             _content_delivered = bool(
                 _sc and getattr(_sc, "final_content_delivered", False)
             )
-            if not _is_empty_sentinel and (_streamed or _previewed or _content_delivered):
+            # Plugin hooks (e.g. transform_llm_output) may have appended content
+            # after streaming finished — when the response was transformed, always
+            # send the final version so the appended content reaches the client.
+            _transformed = bool(response.get("response_transformed"))
+            if not _is_empty_sentinel and not _transformed and (_streamed or _previewed or _content_delivered):
                 logger.info(
                     "Suppressing normal final send for session %s: final delivery already confirmed (streamed=%s previewed=%s content_delivered=%s).",
                     session_key or "?",