diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py
index bb7fbd81146..984934daaac 100644
--- a/litellm/litellm_core_utils/realtime_streaming.py
+++ b/litellm/litellm_core_utils/realtime_streaming.py
@@ -1135,7 +1135,7 @@ class RealTimeStreaming:
             return BackendClose(code=1011, reason="proxy failed while relaying the upstream websocket")
 
     def _backend_refused_session(self, close: BackendClose) -> bool:
-        return close.code != 1000 and not self._backend_sent_frames and not self.messages
+        return close.code != 1000 and not self._backend_sent_frames
 
     async def log_backend_refusal(self, error: Exception) -> None:
         if not self.logging_obj:
diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
index 00addb613c2..09757c35570 100644
--- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
+++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
@@ -3031,12 +3031,12 @@ async def test_session_close_flushes_unbilled_transcription_usage():
     messages before log_messages runs, and never forwarded to the client."""
     from typing import Final
 
-    from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage
+    from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage, RealtimeResponseTypedDict
 
     client_ws: Final = MagicMock()
     client_ws.send_text = AsyncMock()
     backend_ws: Final = MagicMock()
-    backend_ws.recv = AsyncMock(side_effect=ConnectionClosed(None, None))
+    backend_ws.recv = AsyncMock(side_effect=[b'{"serverContent": {}}', ConnectionClosed(None, None)])
     logging_obj: Final = MagicMock()
     logging_obj.async_success_handler = AsyncMock()
     logging_obj.success_handler = MagicMock()
@@ -3048,7 +3048,24 @@ async def test_session_close_flushes_unbilled_transcription_usage():
         "total_tokens": 171,
         "input_token_details": {"text_tokens": 0, "audio_tokens": 153},
     }
+    transcript_frame: Final[RealtimeResponseTypedDict] = {
+        "response": {
+            "type": "conversation.item.input_audio_transcription.completed",
+            "event_id": "event_1",
+            "transcript": "ahoy",
+            "item_id": "item_1",
+            "content_index": 0,
+        },
+        "current_output_item_id": None,
+        "current_response_id": None,
+        "current_delta_chunks": None,
+        "current_conversation_id": None,
+        "current_item_chunks": None,
+        "current_delta_type": None,
+        "session_configuration_request": None,
+    }
     provider_config: Final = MagicMock()
+    provider_config.transform_realtime_response = MagicMock(return_value=transcript_frame)
     provider_config.unbilled_usage_on_session_close = MagicMock(return_value=usage)
 
     streaming: Final = RealTimeStreaming(
@@ -3080,7 +3097,9 @@ async def test_session_close_flushes_unbilled_transcription_usage():
     )
     assert len(flushed) == 1
     assert flushed[0] in logged_snapshots[0]
-    assert not client_ws.send_text.called
+    forwarded: Final = tuple(json.loads(call.args[0]) for call in client_ws.send_text.await_args_list)
+    assert [event.get("transcript") for event in forwarded] == ["ahoy"]
+    assert all("usage" not in event for event in forwarded)
 
 
 @pytest.mark.asyncio
@@ -3246,6 +3265,20 @@ async def test_upstream_refusal_before_any_frame_logs_a_failure_not_a_success():
     assert session.logging.logged_sessions == ()
 
 
+@pytest.mark.asyncio
+async def test_upstream_refusal_after_a_synthetic_session_created_still_logs_a_failure():
+    """LIT-6973: deferred Gemini Live setup stores a synthetic ``session.created`` before
+    the relay starts. It is not an upstream frame, so a refusal after it is still a refusal."""
+    upstream_close: Final = ConnectionClosed(Close(1008, _UPSTREAM_REFUSAL), None)
+    session: Final = _relay_session(_client_ws_that_never_sends(), _backend_ws_closing_with(upstream_close))
+    session.streaming.store_message(json.dumps({"type": "session.created", "session": {"id": "sess_synthetic"}}))
+
+    await session.run()
+
+    assert session.logging.logged_failures == (upstream_close,)
+    assert session.logging.logged_sessions == ()
+
+
 @pytest.mark.asyncio
 async def test_upstream_close_after_relayed_events_still_logs_the_session_as_success():
     client_ws: Final = _client_ws_that_never_sends()