sync: 159 file da Baida98/AI@341f6c21 (2026-08-16 08:37 UTC) [deploy-all]

models/ai_client.py

@@ -452,33 +452,46 @@ class AIClient:
         except Exception as exc:
             _logger.debug("Streaming fleet expansion skipped: %s", type(exc).__name__)
 
-        # Nello streaming proviamo un profilo ruotato per endpoint, poi i fallback runtime.
-        providers = self._execution_pool(providers, "stream")
+        # Un profilo sano per provider: se l’intero pool primario è in rate
+        # limit, il fallback passa automaticamente al provider successivo.
+        providers = self._inter_provider_fallback_pool("stream")
+        attempted: list[str] = []
         for provider in providers:
+            attempted.append(provider.name)
             client = self._client_for(provider)
+            emitted = False
             try:
                 stream = await asyncio.to_thread(
                     client.chat.completions.create,
-                    model=provider.default_model, 
-                    messages=messages, 
-                    temperature=temperature, 
-                    max_tokens=max_tokens, 
+                    model=provider.default_model,
+                    messages=messages,
+                    temperature=temperature,
+                    max_tokens=max_tokens,
                     stream=True,
                 )
                 iterator = iter(stream)
                 while True:
                     chunk = await asyncio.to_thread(next, iterator, None)
-                    if chunk is None: break
+                    if chunk is None:
+                        break
                     if chunk.choices and chunk.choices[0].delta.content:
+                        emitted = True
                         yield chunk.choices[0].delta.content
                 self._record_success(provider)
                 return
             except Exception as e:
                 self._record_failure(provider, e)
-                _logger.warning(f"Streaming fallito su {provider.name}/{provider.profile}: {e}")
+                _logger.warning(
+                    "Streaming fallito su %s/%s (emitted=%s): %s",
+                    provider.name, provider.profile, emitted, e,
+                )
+                # Retry solo prima del primo chunk: dopo output parziale un
+                # retry produrrebbe testo duplicato o una risposta incoerente.
+                if emitted:
+                    raise
                 continue
-        
-        raise ProviderUnavailableError([provider.name for provider in providers])
+
+        raise ProviderUnavailableError(attempted)
 
 
 

tests/test_provider_profile_pool.py

@@ -1,5 +1,6 @@
 import json
 import os
+import types
 import unittest
 from unittest.mock import AsyncMock, patch
 
@@ -104,5 +105,65 @@ class InterProviderFallbackChatTests(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(answer, "fallback answer from healthy provider")
 
 
+class StreamingFallbackTests(unittest.IsolatedAsyncioTestCase):
+    @staticmethod
+    def _chunk(text):
+        return types.SimpleNamespace(
+            choices=[types.SimpleNamespace(
+                delta=types.SimpleNamespace(content=text),
+            )],
+        )
+
+    async def test_stream_retries_next_provider_before_first_chunk(self):
+        client = AIClient()
+        first = ProviderConfig(name="openrouter", api_key="or", base_url="https://openrouter.ai/api/v1", profile="a", purpose="coding", tier=1)
+        second = ProviderConfig(name="groq", api_key="groq", base_url="https://api.groq.com/openai/v1", profile="a", purpose="reasoning", tier=0)
+        client.providers = [first, second]
+
+        class FakeCompletions:
+            def __init__(self, provider):
+                self.provider = provider
+            def create(self, **_kwargs):
+                if self.provider == "openrouter":
+                    raise RuntimeError("HTTP 429 rate limit")
+                return iter([StreamingFallbackTests._chunk("healthy "), StreamingFallbackTests._chunk("stream")])
+
+        def fake_client(provider):
+            return types.SimpleNamespace(chat=types.SimpleNamespace(completions=FakeCompletions(provider.name)))
+
+        with patch.object(client, "_client_for", side_effect=fake_client):
+            output = [part async for part in client.stream_chat([{"role": "user", "content": "hello"}])]
+
+        self.assertEqual(output, ["healthy ", "stream"])
+
+    async def test_stream_does_not_retry_after_partial_output(self):
+        client = AIClient()
+        first = ProviderConfig(name="openrouter", api_key="or", base_url="https://openrouter.ai/api/v1", profile="a", purpose="coding", tier=1)
+        second = ProviderConfig(name="groq", api_key="groq", base_url="https://api.groq.com/openai/v1", profile="a", purpose="reasoning", tier=2)
+        client.providers = [first, second]
+        calls = []
+
+        class FakeCompletions:
+            def __init__(self, provider):
+                self.provider = provider
+            def create(self, **_kwargs):
+                calls.append(self.provider)
+                if self.provider == "openrouter":
+                    def broken_stream():
+                        yield StreamingFallbackTests._chunk("partial")
+                        raise RuntimeError("stream disconnected")
+                    return broken_stream()
+                return iter([StreamingFallbackTests._chunk("should not run")])
+
+        def fake_client(provider):
+            return types.SimpleNamespace(chat=types.SimpleNamespace(completions=FakeCompletions(provider.name)))
+
+        with patch.object(client, "_client_for", side_effect=fake_client):
+            with self.assertRaises(RuntimeError):
+                _ = [part async for part in client.stream_chat([{"role": "user", "content": "hello"}])]
+
+        self.assertEqual(calls, ["openrouter"])
+
+
 if __name__ == "__main__":
     unittest.main()