fix(gateway): skip attachment upload for failed first turns in queued delivery

Adds the failed-result guard the salvage review called for:
_deliver_queued_first_response now takes deliver_media and the queued
follow-up call site passes deliver_media=not _delivery_result.get('failed').
A failed turn still delivers its normalized failure text (pinned by
test_run_agent_sends_normalized_failure_before_queued_followup), but its
attachments are no longer uploaded as if the turn succeeded — mirroring
the completed-turn path's 'not agent_result.get(failed)' guard.
Regression test added.

gateway/run.py

@@ -19945,6 +19945,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         metadata: Optional[Dict[str, Any]] = None,
         event_message_id: Optional[str] = None,
         text_already_delivered: bool = False,
+        deliver_media: bool = True,
     ) -> None:
         """Deliver a queued response using the normal text+attachment split."""
         if not text_already_delivered:
@@ -19956,6 +19957,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     metadata=metadata,
                 )
 
+        # Failed turns still deliver their (normalized failure) text above,
+        # but must not upload attachments as if the turn succeeded — mirrors
+        # the ``not agent_result.get("failed")`` guard on the completed-turn
+        # delivery path.
+        if not deliver_media:
+            return
+
         synthetic_event = MessageEvent(
             text="",
             source=source,
@@ -26364,6 +26372,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                                 metadata=_status_thread_metadata,
                                 event_message_id=event_message_id,
                                 text_already_delivered=_already_streamed,
+                                deliver_media=not _delivery_result.get("failed"),
                             )
                         except Exception as e:
                             logger.warning("Failed to send first response before queued message: %s", e)

tests/gateway/test_tts_media_routing.py

@@ -414,6 +414,44 @@ async def test_queued_followup_delivery_preserves_protected_media_example():
     adapter.send_document.assert_not_awaited()
 
 
+@pytest.mark.asyncio
+async def test_queued_followup_delivery_skips_media_when_turn_failed():
+    """A failed first turn delivers its (failure) text but never uploads
+    attachments as if the turn succeeded — deliver_media=False mirrors the
+    completed-turn path's ``not agent_result.get("failed")`` guard."""
+    event = _event(thread_id="topic-1")
+    runner = object.__new__(GatewayRunner)
+    runner._thread_metadata_for_source = lambda source, anchor=None: {"thread_id": "topic-1"}
+    runner._reply_anchor_for_event = lambda event: event.message_id
+
+    adapter = SimpleNamespace(
+        name="test",
+        extract_media=BasePlatformAdapter.extract_media,
+        extract_images=BasePlatformAdapter.extract_images,
+        extract_local_files=BasePlatformAdapter.extract_local_files,
+        send=AsyncMock(return_value=SendResult(success=True, message_id="text")),
+        send_multiple_images=AsyncMock(return_value=None),
+        send_voice=AsyncMock(return_value=SendResult(success=True, message_id="voice")),
+        send_document=AsyncMock(return_value=SendResult(success=True, message_id="doc")),
+        send_video=AsyncMock(return_value=SendResult(success=True, message_id="video")),
+    )
+
+    await GatewayRunner._deliver_queued_first_response(
+        runner,
+        "The request failed: provider exploded\nMEDIA:/tmp/pricelist.png",
+        source=event.source,
+        adapter=adapter,
+        metadata={"thread_id": "topic-1"},
+        event_message_id=event.message_id,
+        deliver_media=False,
+    )
+
+    adapter.send.assert_awaited_once()
+    adapter.send_multiple_images.assert_not_awaited()
+    adapter.send_document.assert_not_awaited()
+    adapter.send_video.assert_not_awaited()
+
+
 class _QueuedMediaCaptureAdapter(BasePlatformAdapter):
     """Adapter that records text + native image delivery for queued-resend tests."""