pont v73 : guerison du bassin elargie (HTTPError/RuntimeError) + flux SSE gueri

anthropic_proxy.py

@@ -154,6 +154,24 @@ async def _guerir_bassin(request: Request, call_next):
                      "error": {"type": "overloaded_error",
                                "message": "upstream pool exhausted, "
                                           "connection pool rebuilt; retry"}})
+    except (httpx.HTTPError, RuntimeError) as e:
+        # v73 : mesure du 29/08 -- la saturation ne se presente pas toujours en
+        # PoolTimeout. Un RuntimeError("client has been closed") apres une
+        # reconstruction, ou un ReadError sur une connexion recyclee, sortait en
+        # 500 brut sans jamais declencher la guerison. Meme remede, et le type
+        # REEL est journalise pour qu'on ne rediagnostique plus a l'aveugle.
+        import traceback
+        print(f"[bassin?] {type(e).__name__}: {e} sur {request.url.path}",
+              flush=True)
+        traceback.print_exc()
+        await _reconstruire_bassin(generation)
+        return JSONResponse(
+            status_code=503,
+            content={"type": "error",
+                     "error": {"type": "overloaded_error",
+                               "message": "upstream relay failed "
+                                          f"({type(e).__name__}), pool "
+                                          "rebuilt; retry"}})
 
 # Pool EXPLICITE. Mesure 22/08 nuit (5 agents Claude Code, ~150 k de contexte) :
 # 189 flux ouverts pour 100 termines, 45 connexions amont pour 2 clients -- les
@@ -614,6 +632,26 @@ def _sse(event: str, data: dict) -> bytes:
     return f"event: {event}\ndata: {json.dumps(data, separators=(',', ':'))}\n\n".encode()
 
 
+async def _flux_gueri(gen):
+    """v73 : les exceptions nees dans un generateur SSE ne traversent pas le
+    middleware _guerir_bassin. On les attrape ici : evenement `error` au client
+    (sa logique de reprise lit le libelle) et reconstruction du bassin."""
+    generation = _bassin_generation
+    try:
+        async for morceau in gen:
+            yield morceau
+    except (httpx.HTTPError, RuntimeError) as e:
+        import traceback
+        print(f"[bassin?] flux: {type(e).__name__}: {e}", flush=True)
+        traceback.print_exc()
+        asyncio.ensure_future(_reconstruire_bassin(generation))
+        yield _sse("error", {
+            "type": "error",
+            "error": {"type": "overloaded_error",
+                      "message": f"upstream stream failed "
+                                 f"({type(e).__name__}); retry"}})
+
+
 async def stream_anthropic(payload: dict, req_model: str, request: Request | None = None):
     """Traduit le flux OpenAI en evenements Anthropic.
 
@@ -944,7 +982,8 @@ async def messages(request: Request):
             async for morceau in stream_anthropic(payload, req_model, request):
                 yield morceau
 
-        return StreamingResponse(flux_avec_swap(), media_type="text/event-stream")
+        return StreamingResponse(_flux_gueri(flux_avec_swap()),
+                                 media_type="text/event-stream")
     payload["model"] = await _assurer_modele(req_model)
 
     r = await _client.post("/v1/chat/completions", json=payload)
@@ -1086,7 +1125,8 @@ async def oai_chat(request: Request):
             async with _client.stream("POST", "/v1/chat/completions", json=body) as r:
                 async for chunk in r.aiter_raw():
                     yield chunk
-        return StreamingResponse(gen(), media_type="text/event-stream")
+        return StreamingResponse(_flux_gueri(gen()),
+                                 media_type="text/event-stream")
     r = await _client.post("/v1/chat/completions", json=body)
     return JSONResponse(status_code=r.status_code, content=r.json())