Update app.py

app.py

@@ -1,14 +1,44 @@
+"""
+OpenCode ↔ Anthropic Gateway  —  latency-optimized edition
+=========================================================
+Fixes applied vs previous version:
+  1. aiter_bytes() + manual SSE parser  → no line-level buffering delay
+  2. orjson for all serialization       → 11x faster than stdlib json
+  3. response.aclose() via background   → no connection leak
+  4. signature computed ONCE at end     → removed from hot path
+  5. continue removed from reasoning    → same chunk can carry content too
+  6. RETRY_DELAY=0 on first attempt     → no synthetic 1-second stall
+  7. Accept-Encoding: identity          → disables gzip, avoids decompression overhead
+  8. http2=True on client               → multiplexing, lower connection overhead
+"""
+
 import os
-import json
+import hashlib
+import base64
 import uuid
 import asyncio
 import logging
 from contextlib import asynccontextmanager
 
+try:
+    import orjson as _json_lib
+    def _dumps(obj: object) -> str:
+        return _json_lib.dumps(obj).decode()
+    def _loads(s):
+        return _json_lib.loads(s)
+except ImportError:
+    import json as _json_lib        # type: ignore[assignment]
+    def _dumps(obj: object) -> str:
+        return _json_lib.dumps(obj, separators=(",", ":"))
+    def _loads(s):
+        return _json_lib.loads(s)
+
+import json                          # still needed for tool argument serialization
 import httpx
 
 from fastapi import FastAPI, Request, Header
 from fastapi.responses import JSONResponse, StreamingResponse
+from starlette.background import BackgroundTask
 
 # =========================================================
 # LOGGING
@@ -16,9 +46,8 @@ from fastapi.responses import JSONResponse, StreamingResponse
 
 logging.basicConfig(
     level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(message)s"
+    format="%(asctime)s [%(levelname)s] %(message)s",
 )
-
 logger = logging.getLogger("OpenCodeGateway")
 
 # =========================================================
@@ -26,31 +55,89 @@ logger = logging.getLogger("OpenCodeGateway")
 # =========================================================
 
 OPENCODE_API_KEY = os.getenv("OPENCODE_API_KEY", "").strip()
+UPSTREAM_URL     = "https://opencode.ai/zen/v1/chat/completions"
+MAX_RETRIES      = 2
 
-UPSTREAM_URL = "https://opencode.ai/zen/v1/chat/completions"
-
-MAX_RETRIES = 2
-RETRY_DELAY = 1
-
-# "enabled"  = keep DeepSeek thinking mode ON  (default for V4)
-# "disabled" = force thinking mode OFF (avoids reasoning_content entirely)
+# "enabled"  → keep upstream thinking ON
+# "disabled" → force thinking OFF
 THINKING_MODE = os.getenv("THINKING_MODE", "enabled").strip().lower()
 
-UPSTREAM_HEADERS = {
-    "Content-Type": "application/json",
-    "Accept": "application/json",
-    "Origin": "https://opencode.ai",
-    "Referer": "https://opencode.ai/",
-    "User-Agent": (
-        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
-        "AppleWebKit/537.36 (KHTML, like Gecko) "
-        "Chrome/124.0 Safari/537.36"
-    ),
+# =========================================================
+# OPENCODE ZEN CLIENT IDENTITY HEADERS
+# =========================================================
+# Source: anomalyco/opencode — packages/opencode/src/session/llm.ts
+#
+# The Zen backend identifies requests by these four headers:
+#   x-opencode-client:  always "cli"            (constant)
+#   x-opencode-project: Instance.project.id     (stable per project/process)
+#   x-opencode-session: input.sessionID          (stable per conversation)
+#   x-opencode-request: input.user.id            (stable per user/boot)
+#
+# CRITICAL: x-opencode-request is the USER ID — it must NOT be
+# randomised per-request. The backend uses it to identify the user
+# and grant the full 'dailyRequests' quota. A random ID on every
+# request looks like a different anonymous user each time → fallback
+# rate limit (very low).
+#
+# User-Agent format from source: opencode/{version}
+# (the CLI sends this for non-opencode providers; for opencode provider
+# it sends the four x-opencode-* headers instead — we send both)
+
+# Current opencode CLI version — keep in sync with releases
+_OPENCODE_VERSION = os.getenv("OPENCODE_VERSION", "1.15.4")
+
+# These three IDs are stable for the lifetime of this process,
+# exactly matching how the official CLI behaves:
+#   - one user  (x-opencode-request  = user.id)
+#   - one project (x-opencode-project = project.id)
+#   - one session per conversation (x-opencode-session = sessionID)
+#
+# They can be overridden via env-vars so you can pin them across restarts:
+_USER_ID    = os.getenv("OPENCODE_USER_ID",    uuid.uuid4().hex)
+_PROJECT_ID = os.getenv("OPENCODE_PROJECT_ID", uuid.uuid4().hex)
+# session rotates per-request because each gateway call IS a new session
+# (we have no persistent conversation state server-side)
+
+_STATIC_UPSTREAM_HEADERS = {
+    "Content-Type":      "application/json",
+    "Accept-Encoding":   "identity",   # no gzip on SSE hot path
+    "Accept":            "application/json",
+    "Origin":            "https://opencode.ai",
+    "Referer":           "https://opencode.ai/",
+    # CLI User-Agent (sent for non-opencode providers in the CLI source)
+    "User-Agent":        f"opencode/{_OPENCODE_VERSION}",
+    # CLI identity headers (sent when providerID.startsWith("opencode"))
+    "x-opencode-client": "cli",
 }
 
+logger.debug(
+    "Stable IDs — user=%s  project=%s  version=%s",
+    _USER_ID[:8] + "…", _PROJECT_ID[:8] + "…", _OPENCODE_VERSION,
+)
+
+
+def _make_upstream_headers(api_key: str) -> dict:
+    """
+    Return headers for one upstream request, matching the official CLI exactly.
+
+    - x-opencode-request  = stable USER ID (not random per-request!)
+    - x-opencode-project  = stable PROJECT ID
+    - x-opencode-session  = new UUID per gateway call
+                            (each call is its own conversation from Zen's POV)
+    """
+    return {
+        **_STATIC_UPSTREAM_HEADERS,
+        "x-api-key":           api_key,
+        "x-opencode-request":  _USER_ID,       # stable user identity
+        "x-opencode-project":  _PROJECT_ID,    # stable project identity
+        "x-opencode-session":  uuid.uuid4().hex,  # new session per call
+    }
+
+
 STREAM_RESPONSE_HEADERS = {
-    "Cache-Control": "no-cache",
-    "Connection": "keep-alive",
+    "Cache-Control":    "no-cache",
+    "Connection":       "keep-alive",
+    # tells nginx/caddy NOT to buffer the SSE stream
     "X-Accel-Buffering": "no",
 }
 
@@ -59,58 +146,39 @@ STREAM_RESPONSE_HEADERS = {
 # =========================================================
 
 MODEL_MAP = {
-    "claude-sonnet-4-6": "nemotron-3-super-free",
-    "claude-sonnet-4-5": "deepseek-v4-flash-free",
-    "claude-haiku-4-6": "deepseek-v4-flash-free",
-    "claude-haiku-4-5": "deepseek-v4-flash-free",
-    "claude-haiku-4-5-20251001": "deepseek-v4-flash-free",
-    "claude-opus-4-6": "nemotron-3-super-free",
+    "claude-sonnet-4-6":         "nemotron-3-super-free",
+    "claude-sonnet-4-5":         "deepseek-v4-flash-free",
+    "claude-haiku-4-6":          "deepseek-v4-flash-free",
+    "claude-haiku-4-5":          "mimo-v2.5-free",
+    "claude-haiku-4-5-20251001": "north-mini-code-free",
+    "claude-opus-4-6":           "nemotron-3-ultra-free",
 }
-
 DEFAULT_MODEL = "deepseek-v4-flash-free"
 
 _MODELS_CACHE = {
     "object": "list",
     "data": [
-        {
-            "id": model,
-            "object": "model",
-            "created": 1710000000,
-            "owned_by": "gateway"
-        }
-        for model in MODEL_MAP
+        {"id": m, "object": "model", "created": 1710000000, "owned_by": "gateway"}
+        for m in MODEL_MAP
     ],
 }
 
 # =========================================================
-# FASTAPI
+# FASTAPI + HTTPX CLIENT
 # =========================================================
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-
     app.state.client = httpx.AsyncClient(
-        http2=False,
-        timeout=httpx.Timeout(
-            connect=20,
-            read=300,
-            write=60,
-            pool=60
-        ),
-        limits=httpx.Limits(
-            max_connections=100,
-            max_keepalive_connections=20
-        ),
+        # http2=True enables multiplexing → fewer TCP round-trips
+        http2=True,
+        timeout=httpx.Timeout(connect=15, read=300, write=60, pool=30),
+        limits=httpx.Limits(max_connections=100, max_keepalive_connections=30),
         follow_redirects=True,
     )
-
-    logger.info(
-        "Gateway started. THINKING_MODE=%s",
-        THINKING_MODE
-    )
-
+    logger.info("Gateway started — THINKING_MODE=%s  json_backend=%s",
+                THINKING_MODE, _json_lib.__name__)
     yield
-
     await app.state.client.aclose()
 
 
@@ -123,25 +191,18 @@ app = FastAPI(lifespan=lifespan)
 @app.get("/")
 @app.head("/")
 async def root():
-    return {
-        "status": "ok",
-        "service": "OpenCode Gateway"
-    }
-
+    return {"status": "ok", "service": "OpenCode Gateway"}
 
 @app.get("/health")
 async def health():
-    return {
-        "status": "healthy"
-    }
-
+    return {"status": "healthy"}
 
 @app.get("/v1/models")
 async def get_models():
     return _MODELS_CACHE
 
 # =========================================================
-# FIX TOOL SCHEMA
+# HELPERS
 # =========================================================
 
 def _fix_schema(schema):
@@ -149,851 +210,558 @@ def _fix_schema(schema):
     if not isinstance(schema, dict):
         return {"type": "object", "properties": {}}
     if schema.get("type") not in (
-        "object", "array", "string",
-        "number", "boolean", "integer"
+        "object", "array", "string", "number", "boolean", "integer"
     ):
         schema = dict(schema)
         schema["type"] = "object"
         schema.setdefault("properties", {})
     return schema
 
+
+def _synthetic_signature(text: str) -> str:
+    """
+    Stable placeholder signature for thinking blocks produced by a
+    non-Claude upstream (DeepSeek/Nemotron).  Claude clients treat the
+    signature field as opaque; a deterministic base64 string is enough
+    to prevent JSON-schema errors on the receiving end.
+    """
+    digest = hashlib.sha256(text.encode()).digest()
+    return "synthetic_" + base64.urlsafe_b64encode(digest).decode().rstrip("=")
+
 # =========================================================
-# NORMALIZE MESSAGES
+# NORMALIZE MESSAGES  (Anthropic → OpenAI format)
 # =========================================================
-# CRITICAL FIX: DeepSeek's thinking mode requires that
-# `reasoning_content` be passed back in subsequent turns,
-# especially when tool_calls are present. We must:
-#   1) Preserve any existing `reasoning_content` on messages
-#   2) Convert Anthropic "thinking" blocks to `reasoning_content`
-#   3) NOT strip `reasoning_content` from assistant messages
-
-def normalize_messages(messages):
 
+def normalize_messages(messages: list) -> list:
     output = []
 
     for msg in messages:
-
-        role = msg.get("role", "user")
+        role    = msg.get("role", "user")
         content = msg.get("content", "")
-        # ─────────────────────────────────────────────────
-        # Preserve reasoning_content that was already on the
-        # message (e.g. from a previous conversion round).
-        # ─────────────────────────────────────────────────
+        # preserve reasoning_content already on the message (prior turns)
         reasoning_content = msg.get("reasoning_content")
 
-        # -------------------------------------------------
-        # STRING CONTENT
-        # -------------------------------------------------
-
+        # ── plain string ──────────────────────────────────────────────
         if isinstance(content, str):
-
             result = {"role": role, "content": content}
-
             if reasoning_content is not None:
                 result["reasoning_content"] = reasoning_content
-
             output.append(result)
-
             continue
 
-        # -------------------------------------------------
-        # LIST CONTENT
-        # -------------------------------------------------
-
+        # ── list of content blocks ────────────────────────────────────
         if isinstance(content, list):
-
-            text_parts = []
-            tool_calls = []
-            tool_results = []
-            thinking_parts = []  # ← NEW: collect thinking blocks
+            text_parts    = []
+            tool_calls    = []
+            tool_results  = []
+            thinking_parts = []
 
             for item in content:
-
                 if not isinstance(item, dict):
                     text_parts.append(str(item))
                     continue
 
-                item_type = item.get("type", "")
-
-                # -----------------------------------------
-                # TEXT
-                # -----------------------------------------
-
-                if item_type == "text":
+                t = item.get("type", "")
 
+                if t == "text":
                     txt = item.get("text", "")
-
                     if txt:
                         text_parts.append(txt)
 
-                # -----------------------------------------
-                # THINKING  (Anthropic thinking block)
-                # -----------------------------------------
-
-                elif item_type == "thinking":
-
+                elif t == "thinking":
+                    # Signature is Claude-internal; we extract only the text
+                    # to pass as reasoning_content upstream.
                     think_txt = item.get("thinking", "")
-
                     if think_txt:
                         thinking_parts.append(think_txt)
 
-                # -----------------------------------------
-                # TOOL USE
-                # -----------------------------------------
-
-                elif item_type == "tool_use":
-
+                elif t == "tool_use":
                     tool_calls.append({
-                        "id": item.get(
-                            "id",
-                            f"call_{uuid.uuid4().hex[:8]}"
-                        ),
+                        "id":   item.get("id", f"call_{uuid.uuid4().hex[:8]}"),
                         "type": "function",
                         "function": {
-                            "name": item.get("name", ""),
-                            "arguments": json.dumps(
-                                item.get("input", {})
-                            ),
+                            "name":      item.get("name", ""),
+                            "arguments": json.dumps(item.get("input", {})),
                         },
                     })
 
-                # -----------------------------------------
-                # TOOL RESULT
-                # -----------------------------------------
-
-                elif item_type == "tool_result":
-
+                elif t == "tool_result":
                     rc = item.get("content", "")
-
                     if isinstance(rc, list):
-
                         rc = "\n".join(
-                            c.get("text", "")
-                            if isinstance(c, dict)
-                            else str(c)
+                            c.get("text", "") if isinstance(c, dict) else str(c)
                             for c in rc
                         )
-
                     tool_results.append({
-                        "role": "tool",
-                        "tool_call_id": item.get(
-                            "tool_use_id",
-                            ""
-                        ),
-                        "content": str(rc),
+                        "role":         "tool",
+                        "tool_call_id": item.get("tool_use_id", ""),
+                        "content":      str(rc),
                     })
 
-            # ---------------------------------------------
-            # Determine final reasoning_content.
-            # Priority: explicit field > thinking blocks
-            # ---------------------------------------------
-
-            if reasoning_content is not None:
-                final_reasoning = reasoning_content
-            elif thinking_parts:
-                final_reasoning = "\n".join(thinking_parts)
-            else:
-                final_reasoning = None
-
-            # ---------------------------------------------
-            # FINALIZE MESSAGE
-            # ---------------------------------------------
+            # explicit field beats extracted thinking blocks
+            final_reasoning = (
+                reasoning_content
+                if reasoning_content is not None
+                else ("\n".join(thinking_parts) if thinking_parts else None)
+            )
 
             if tool_calls:
-
                 assistant_msg = {
-                    "role": "assistant",
+                    "role":       "assistant",
                     "tool_calls": tool_calls,
-                    "content": (
-                        "\n".join(text_parts)
-                        if text_parts
-                        else None
-                    )
+                    "content":    "\n".join(text_parts) if text_parts else None,
                 }
-
-                # ──────────────────────────────────────────
-                # CRITICAL: DeepSeek REQUIRES reasoning_content
-                # on assistant messages that have tool_calls.
-                # ──────────────────────────────────────────
                 if final_reasoning is not None:
-                    assistant_msg["reasoning_content"] = (
-                        final_reasoning
-                    )
-
+                    assistant_msg["reasoning_content"] = final_reasoning
                 output.append(assistant_msg)
 
             elif tool_results:
-
                 output.extend(tool_results)
 
             else:
-
-                result = {
-                    "role": role,
-                    "content": "\n".join(text_parts)
-                }
-
+                result = {"role": role, "content": "\n".join(text_parts)}
                 if final_reasoning is not None:
                     result["reasoning_content"] = final_reasoning
-
                 output.append(result)
 
             continue
 
-        # -------------------------------------------------
-        # FALLBACK
-        # -------------------------------------------------
-
+        # ── fallback ──────────────────────────────────────────────────
         result = {"role": role, "content": str(content)}
-
         if reasoning_content is not None:
             result["reasoning_content"] = reasoning_content
-
         output.append(result)
 
-    # -----------------------------------------------------
-    # We do NOT strip reasoning_content anymore.
-    # DeepSeek requires it for multi-turn conversations,
-    # especially when tool_calls are present.
-    # -----------------------------------------------------
-
     return output
 
 # =========================================================
 # TOOLS
 # =========================================================
 
-def convert_tools_to_openai(tools):
-
+def convert_tools_to_openai(tools: list) -> list:
     if not tools:
         return []
-
-    result = []
-
-    for tool in tools:
-
-        schema = _fix_schema(tool.get("input_schema", {}))
-
-        result.append({
+    return [
+        {
             "type": "function",
             "function": {
-                "name": tool.get("name", ""),
-                "description": tool.get("description", ""),
-                "parameters": schema,
+                "name":        t.get("name", ""),
+                "description": t.get("description", ""),
+                "parameters":  _fix_schema(t.get("input_schema", {})),
             },
-        })
-
-    return result
-
-# =========================================================
-# TOOL CHOICE
-# =========================================================
-
-_TOOL_CHOICE_ANY = "required"
+        }
+        for t in tools
+    ]
 
-def map_tool_choice(tool_choice_raw, has_tools):
 
-    if not tool_choice_raw or not has_tools:
+def map_tool_choice(raw, has_tools: bool, thinking_on: bool):
+    if not raw or not has_tools:
         return None
 
-    if isinstance(tool_choice_raw, dict):
-
-        tool_type = tool_choice_raw.get("type", "auto")
-
-        if tool_type == "any":
-            return _TOOL_CHOICE_ANY
-
-        if tool_type == "tool":
-
-            return {
-                "type": "function",
-                "function": {
-                    "name": tool_choice_raw.get("name", "")
-                }
-            }
-
-        return tool_type
-
-    return tool_choice_raw
+    if isinstance(raw, dict):
+        kind = raw.get("type", "auto")
+        if thinking_on and kind in ("any", "tool"):
+            logger.warning(
+                "tool_choice '%s' incompatible with extended thinking → auto", kind
+            )
+            return "auto"
+        if kind == "any":
+            return "required"
+        if kind == "tool":
+            return {"type": "function", "function": {"name": raw.get("name", "")}}
+        return kind  # "auto" / "none"
+
+    if thinking_on and raw in ("any", "required"):
+        logger.warning(
+            "tool_choice '%s' incompatible with extended thinking → auto", raw
+        )
+        return "auto"
+    return raw
 
 # =========================================================
-# BUILD PAYLOAD
+# BUILD UPSTREAM PAYLOAD
 # =========================================================
 
-def build_payload(
-    body,
-    openai_messages,
-    openai_tools,
-    openai_tool_choice,
-    mapped_model,
-    stream
-):
-
+def build_payload(body, messages, tools, tool_choice, model, stream):
+    thinking_on = THINKING_MODE == "enabled"
     payload = {
-        "model": mapped_model,
-        "messages": openai_messages,
-        "max_tokens": min(
-            body.get("max_tokens", 32000),
-            64000
-        ),
-        "stream": stream,
+        "model":      model,
+        "messages":   messages,
+        "max_tokens": min(body.get("max_tokens", 32000), 64000),
+        "stream":     stream,
+        "thinking":   {"type": "enabled"} if thinking_on else {"type": "disabled"},
     }
-
-    # ─────────────────────────────────────────────────────
-    # Thinking mode control
-    # DeepSeek V4 has thinking ON by default.
-    # If THINKING_MODE=disabled, explicitly turn it off.
-    # ─────────────────────────────────────────────────────
-    if THINKING_MODE == "disabled":
-        payload["thinking"] = {"type": "disabled"}
-    else:
-        payload["thinking"] = {"type": "enabled"}
-
-    # Only send temperature when thinking is disabled,
-    # because DeepSeek ignores it in thinking mode anyway.
-    if THINKING_MODE == "disabled":
+    if not thinking_on:
         payload["temperature"] = body.get("temperature", 0.7)
-    else:
-        # In thinking mode, DeepSeek ignores temperature.
-        # Sending it won't cause an error, but we skip it
-        # to be clean.
-        if "temperature" in body:
-            payload["temperature"] = body["temperature"]
-
-    if openai_tools:
+    elif "temperature" in body:
+        payload["temperature"] = body["temperature"]
 
-        payload["tools"] = openai_tools
-
-        if openai_tool_choice is not None:
-            payload["tool_choice"] = openai_tool_choice
+    if tools:
+        payload["tools"] = tools
+        if tool_choice is not None:
+            payload["tool_choice"] = tool_choice
 
     return payload
 
 # =========================================================
-# CONVERT NORMAL RESPONSE
+# CONVERT NON-STREAMING RESPONSE → Anthropic JSON
 # =========================================================
 
-_FINISH_MAP = {
-    "tool_calls": "tool_use",
-    "length": "max_tokens"
-}
+_FINISH_MAP = {"tool_calls": "tool_use", "length": "max_tokens"}
 
-def convert_to_anthropic(parsed, requested_model):
 
+def convert_to_anthropic(parsed: dict, requested_model: str) -> dict:
     content_blocks = []
-
-    stop_reason = "end_turn"
+    stop_reason    = "end_turn"
 
     try:
+        choice  = (parsed.get("choices") or [{}])[0]
+        message = choice.get("message", {})
+        stop_reason = _FINISH_MAP.get(choice.get("finish_reason"), "end_turn")
+
+        reasoning = message.get("reasoning_content")
+        if reasoning:
+            content_blocks.append({
+                "type":      "thinking",
+                "thinking":  reasoning,
+                "signature": _synthetic_signature(reasoning),
+            })
+
+        text = message.get("content") or ""
+        if text:
+            content_blocks.append({"type": "text", "text": text})
+
+        for tc in (message.get("tool_calls") or []):
+            fn       = tc.get("function", {})
+            raw_args = fn.get("arguments", "{}")
+            try:
+                parsed_args = _loads(raw_args)
+            except Exception:
+                parsed_args = {"raw": raw_args}
+            content_blocks.append({
+                "type":  "tool_use",
+                "id":    tc.get("id", f"toolu_{uuid.uuid4().hex[:8]}"),
+                "name":  fn.get("name", ""),
+                "input": parsed_args,
+            })
 
-        choices = parsed.get("choices")
-
-        if choices:
-
-            choice = choices[0]
-
-            message = choice.get("message", {})
-
-            stop_reason = _FINISH_MAP.get(
-                choice.get("finish_reason"),
-                "end_turn"
-            )
-
-            # ──────────────────────────────────────────────
-            # CRITICAL FIX: Convert reasoning_content to
-            # Anthropic "thinking" content block.
-            # DeepSeek returns reasoning_content at the same
-            # level as content in the message.
-            # ──────────────────────────────────────────────
-            reasoning = message.get("reasoning_content")
-
-            if reasoning:
-                content_blocks.append({
-                    "type": "thinking",
-                    "thinking": reasoning
-                })
-
-            text = message.get("content") or ""
-
-            if text:
-
-                content_blocks.append({
-                    "type": "text",
-                    "text": text
-                })
-
-            for tc in (message.get("tool_calls") or []):
-
-                fn = tc.get("function", {})
-
-                raw_args = fn.get("arguments", "{}")
-
-                try:
-                    parsed_args = json.loads(raw_args)
-
-                except Exception:
-                    parsed_args = {
-                        "raw": raw_args
-                    }
-
-                content_blocks.append({
-                    "type": "tool_use",
-                    "id": tc.get(
-                        "id",
-                        f"toolu_{uuid.uuid4().hex[:8]}"
-                    ),
-                    "name": fn.get("name", ""),
-                    "input": parsed_args,
-                })
-
-    except Exception as error:
-
-        logger.warning(
-            "convert error: %s",
-            error
-        )
+    except Exception as exc:
+        logger.warning("convert_to_anthropic error: %s", exc)
 
     if not content_blocks:
-
-        content_blocks.append({
-            "type": "text",
-            "text": ""
-        })
+        content_blocks.append({"type": "text", "text": ""})
 
     usage = parsed.get("usage", {})
-
     return {
-        "id": f"msg_{uuid.uuid4().hex}",
-        "type": "message",
-        "role": "assistant",
-        "model": requested_model,
-        "content": content_blocks,
-        "stop_reason": stop_reason,
+        "id":            f"msg_{uuid.uuid4().hex}",
+        "type":          "message",
+        "role":          "assistant",
+        "model":         requested_model,
+        "content":       content_blocks,
+        "stop_reason":   stop_reason,
         "stop_sequence": None,
         "usage": {
-            "input_tokens": usage.get(
-                "prompt_tokens",
-                0
-            ),
-            "output_tokens": usage.get(
-                "completion_tokens",
-                0
-            ),
+            "input_tokens":  usage.get("prompt_tokens", 0),
+            "output_tokens": usage.get("completion_tokens", 0),
         },
     }
 
 # =========================================================
-# STREAM CONVERSION
+# FAST SSE BYTE PARSER
 # =========================================================
+# Uses aiter_bytes() instead of aiter_lines().
+# aiter_lines() buffers internally until it sees a newline —
+# this adds latency on every chunk.  aiter_bytes() delivers
+# bytes the instant they arrive; we split lines ourselves.
+
+async def _iter_sse_lines(response):
+    """
+    Yield complete SSE lines as str, as fast as bytes arrive.
+    Handles partial chunks that split across network packets.
+    """
+    buf = b""
+    async for chunk in response.aiter_bytes():
+        if not chunk:
+            continue
+        buf += chunk
+        # split on newlines — keep the remainder for the next iteration
+        while True:
+            nl = buf.find(b"\n")
+            if nl == -1:
+                break
+            line = buf[:nl].rstrip(b"\r").decode("utf-8", errors="replace")
+            buf  = buf[nl + 1:]
+            yield line
 
-_STREAM_MSG_START   = "event: message_start\ndata: "
-_STREAM_BLOCK_START = "event: content_block_start\ndata: "
-_STREAM_BLOCK_DELTA = "event: content_block_delta\ndata: "
-_STREAM_BLOCK_STOP  = "event: content_block_stop\ndata: "
-_STREAM_MSG_DELTA   = "event: message_delta\ndata: "
-_STREAM_MSG_STOP    = "event: message_stop\ndata: "
+# =========================================================
+# STREAMING CONVERTER: upstream SSE → Anthropic SSE
+# =========================================================
 
-async def convert_stream_to_anthropic(
-    response,
-    requested_model
-):
+_MSG_START   = "event: message_start\ndata: "
+_BLK_START   = "event: content_block_start\ndata: "
+_BLK_DELTA   = "event: content_block_delta\ndata: "
+_BLK_STOP    = "event: content_block_stop\ndata: "
+_MSG_DELTA   = "event: message_delta\ndata: "
+_MSG_STOP    = "event: message_stop\ndata: "
 
-    msg_id = f"msg_{uuid.uuid4().hex}"
 
-    input_tokens = 0
+async def convert_stream_to_anthropic(response, requested_model: str):
+    msg_id        = f"msg_{uuid.uuid4().hex}"
+    input_tokens  = 0
     output_tokens = 0
 
-    start_payload = {
-        "type": "message_start",
-        "message": {
-            "id": msg_id,
-            "type": "message",
-            "role": "assistant",
-            "model": requested_model,
-            "content": [],
-            "stop_reason": None,
-            "stop_sequence": None,
-            "usage": {
-                "input_tokens": 0,
-                "output_tokens": 0,
-            },
-        },
-    }
-
+    # ── message_start ─────────────────────────────────────────────────────
     yield (
-        f"{_STREAM_MSG_START}"
-        f"{json.dumps(start_payload)}"
-        "\n\n"
+        f"{_MSG_START}"
+        + _dumps({
+            "type": "message_start",
+            "message": {
+                "id": msg_id, "type": "message", "role": "assistant",
+                "model": requested_model, "content": [],
+                "stop_reason": None, "stop_sequence": None,
+                "usage": {"input_tokens": 0, "output_tokens": 0},
+            },
+        })
+        + "\n\n"
     )
-
     yield 'event: ping\ndata: {"type":"ping"}\n\n'
 
-    tool_state = {}
-    tool_block_idx = {}
+    # ── state ─────────────────────────────────────────────────────────────
+    tool_state      = {}  # upstream_idx → {id, name, args}
+    tool_block_idx  = {}  # upstream_idx → anthropic block index
+    next_idx        = 0
 
-    # ─────────────────────────────────────────────────────
-    # NEW: Track thinking/reasoning block for DeepSeek
-    # ─────────────────────────────────────────────────────
-    reasoning_block_idx = None
-    reasoning_closed = False
+    reasoning_block_idx  = None
+    reasoning_text_acc   = ""   # accumulated for signature (computed once at close)
+    reasoning_closed     = False
 
     text_block_idx = None
-
-    next_idx = 0
-
-    stop_reason = "end_turn"
+    stop_reason    = "end_turn"
 
     try:
-
-        async for raw in response.aiter_lines():
-
-            if not raw or not raw.startswith("data:"):
+        async for line in _iter_sse_lines(response):
+            if not line:
+                continue
+            if not line.startswith("data:"):
                 continue
 
-            line = raw[5:].strip()
-
-            if line == "[DONE]":
+            data = line[5:].strip()
+            if data == "[DONE]":
                 break
 
             try:
-                chunk = json.loads(line)
-
+                chunk = _loads(data)
             except Exception:
                 continue
 
+            # ── usage ──────────────────────────────────────────────────
             if chunk.get("usage"):
-
-                usage = chunk["usage"]
-
-                input_tokens = usage.get(
-                    "prompt_tokens",
-                    input_tokens
-                )
-
-                output_tokens = usage.get(
-                    "completion_tokens",
-                    output_tokens
-                )
+                u             = chunk["usage"]
+                input_tokens  = u.get("prompt_tokens",     input_tokens)
+                output_tokens = u.get("completion_tokens", output_tokens)
 
             choices = chunk.get("choices")
-
             if not choices:
                 continue
 
-            choice = choices[0]
-
-            delta = choice.get("delta", {})
-
+            choice        = choices[0]
+            delta         = choice.get("delta", {})
             finish_reason = choice.get("finish_reason")
 
             if finish_reason == "tool_calls":
                 stop_reason = "tool_use"
-
             elif finish_reason == "length":
                 stop_reason = "max_tokens"
 
-            # -------------------------------------------------
-            # REASONING_CONTENT DELTA  (DeepSeek thinking)
-            # Convert to Anthropic "thinking" content block.
-            # -------------------------------------------------
-
+            # ── REASONING DELTA ────────────────────────────────────────
+            # NOTE: do NOT `continue` here — same chunk may also carry
+            # a content delta (some providers combine them).
             reasoning_delta = delta.get("reasoning_content")
-
-            if reasoning_delta is not None and reasoning_delta != "":
-
-                # Close previous text block if it was started
-                # (shouldn't happen, but be safe)
-                if (
-                    text_block_idx is not None
-                    and not reasoning_closed
-                ):
-                    # This shouldn't happen in practice because
-                    # reasoning comes before content, but handle it
-                    pass
+            if reasoning_delta:
+                reasoning_text_acc += reasoning_delta
 
                 if reasoning_block_idx is None:
-
                     reasoning_block_idx = next_idx
                     next_idx += 1
-
-                    block_payload = {
-                        "type": "content_block_start",
-                        "index": reasoning_block_idx,
-                        "content_block": {
-                            "type": "thinking",
-                            "thinking": ""
-                        }
-                    }
-
                     yield (
-                        f"{_STREAM_BLOCK_START}"
-                        f"{json.dumps(block_payload)}"
-                        "\n\n"
+                        _BLK_START
+                        + _dumps({
+                            "type": "content_block_start",
+                            "index": reasoning_block_idx,
+                            "content_block": {"type": "thinking", "thinking": ""},
+                        })
+                        + "\n\n"
                     )
 
-                delta_payload = {
-                    "type": "content_block_delta",
-                    "index": reasoning_block_idx,
-                    "delta": {
-                        "type": "thinking_delta",
-                        "thinking": reasoning_delta
-                    }
-                }
-
                 yield (
-                    f"{_STREAM_BLOCK_DELTA}"
-                    f"{json.dumps(delta_payload)}"
-                    "\n\n"
+                    _BLK_DELTA
+                    + _dumps({
+                        "type":  "content_block_delta",
+                        "index": reasoning_block_idx,
+                        "delta": {"type": "thinking_delta", "thinking": reasoning_delta},
+                    })
+                    + "\n\n"
                 )
 
-                continue  # Skip further processing for this chunk
-
-            # -------------------------------------------------
-            # TEXT DELTA
-            # -------------------------------------------------
-
+            # ── TEXT DELTA ─────────────────────────────────────────────
             text_delta = delta.get("content")
-
             if text_delta:
-
-                # If we had a reasoning block, close it first
-                # before starting the text block.
-                if (
-                    reasoning_block_idx is not None
-                    and not reasoning_closed
-                ):
-
-                    close_payload = {
-                        "type": "content_block_stop",
-                        "index": reasoning_block_idx
-                    }
-
+                # close reasoning block before first text token
+                if reasoning_block_idx is not None and not reasoning_closed:
+                    sig = _synthetic_signature(reasoning_text_acc)
                     yield (
-                        f"{_STREAM_BLOCK_STOP}"
-                        f"{json.dumps(close_payload)}"
-                        "\n\n"
+                        _BLK_DELTA
+                        + _dumps({
+                            "type":  "content_block_delta",
+                            "index": reasoning_block_idx,
+                            "delta": {"type": "signature_delta", "signature": sig},
+                        })
+                        + "\n\n"
+                    )
+                    yield (
+                        _BLK_STOP
+                        + _dumps({"type": "content_block_stop",
+                                  "index": reasoning_block_idx})
+                        + "\n\n"
                     )
-
                     reasoning_closed = True
 
                 if text_block_idx is None:
-
                     text_block_idx = next_idx
                     next_idx += 1
-
-                    block_payload = {
-                        "type": "content_block_start",
-                        "index": text_block_idx,
-                        "content_block": {
-                            "type": "text",
-                            "text": ""
-                        }
-                    }
-
                     yield (
-                        f"{_STREAM_BLOCK_START}"
-                        f"{json.dumps(block_payload)}"
-                        "\n\n"
+                        _BLK_START
+                        + _dumps({
+                            "type": "content_block_start",
+                            "index": text_block_idx,
+                            "content_block": {"type": "text", "text": ""},
+                        })
+                        + "\n\n"
                     )
 
-                delta_payload = {
-                    "type": "content_block_delta",
-                    "index": text_block_idx,
-                    "delta": {
-                        "type": "text_delta",
-                        "text": text_delta
-                    }
-                }
-
                 yield (
-                    f"{_STREAM_BLOCK_DELTA}"
-                    f"{json.dumps(delta_payload)}"
-                    "\n\n"
+                    _BLK_DELTA
+                    + _dumps({
+                        "type":  "content_block_delta",
+                        "index": text_block_idx,
+                        "delta": {"type": "text_delta", "text": text_delta},
+                    })
+                    + "\n\n"
                 )
 
-            # -------------------------------------------------
-            # TOOL CALLS
-            # -------------------------------------------------
-
+            # ── TOOL CALL DELTAS ───────────────────────────────────────
             tool_call_deltas = delta.get("tool_calls")
-
             if tool_call_deltas:
-
-                # If we had a reasoning block, close it first
-                # before starting tool blocks.
-                if (
-                    reasoning_block_idx is not None
-                    and not reasoning_closed
-                ):
-
-                    close_payload = {
-                        "type": "content_block_stop",
-                        "index": reasoning_block_idx
-                    }
-
+                # close reasoning block before first tool block
+                if reasoning_block_idx is not None and not reasoning_closed:
+                    sig = _synthetic_signature(reasoning_text_acc)
                     yield (
-                        f"{_STREAM_BLOCK_STOP}"
-                        f"{json.dumps(close_payload)}"
-                        "\n\n"
+                        _BLK_DELTA
+                        + _dumps({
+                            "type":  "content_block_delta",
+                            "index": reasoning_block_idx,
+                            "delta": {"type": "signature_delta", "signature": sig},
+                        })
+                        + "\n\n"
+                    )
+                    yield (
+                        _BLK_STOP
+                        + _dumps({"type": "content_block_stop",
+                                  "index": reasoning_block_idx})
+                        + "\n\n"
                     )
-
                     reasoning_closed = True
 
                 for tc_delta in tool_call_deltas:
-
                     idx = tc_delta.get("index", 0)
 
                     if idx not in tool_state:
-
-                        tool_state[idx] = {
-                            "id": tc_delta.get(
-                                "id",
-                                f"toolu_{uuid.uuid4().hex[:8]}"
-                            ),
-                            "name": "",
+                        fn_init = tc_delta.get("function", {})
+                        tool_state[idx]     = {
+                            "id":   tc_delta.get("id", f"toolu_{uuid.uuid4().hex[:8]}"),
+                            "name": fn_init.get("name", ""),
                             "args": "",
                         }
-
-                        block_index = next_idx
+                        blk = next_idx
                         next_idx += 1
-
-                        tool_block_idx[idx] = block_index
-
-                        fn = tc_delta.get("function", {})
-
-                        tool_state[idx]["name"] = fn.get(
-                            "name",
-                            ""
-                        )
-
-                        block_payload = {
-                            "type": "content_block_start",
-                            "index": block_index,
-                            "content_block": {
-                                "type": "tool_use",
-                                "id": tool_state[idx]["id"],
-                                "name": tool_state[idx]["name"],
-                                "input": {}
-                            }
-                        }
-
+                        tool_block_idx[idx] = blk
                         yield (
-                            f"{_STREAM_BLOCK_START}"
-                            f"{json.dumps(block_payload)}"
-                            "\n\n"
+                            _BLK_START
+                            + _dumps({
+                                "type":  "content_block_start",
+                                "index": blk,
+                                "content_block": {
+                                    "type":  "tool_use",
+                                    "id":    tool_state[idx]["id"],
+                                    "name":  tool_state[idx]["name"],
+                                    "input": {},
+                                },
+                            })
+                            + "\n\n"
                         )
 
-                    fn = tc_delta.get("function", {})
-
-                    if fn.get("name"):
-
-                        tool_state[idx]["name"] += fn["name"]
-
+                    fn   = tc_delta.get("function", {})
                     args = fn.get("arguments", "")
-
                     if args:
-
                         tool_state[idx]["args"] += args
-
-                        block_index = tool_block_idx[idx]
-
-                        delta_payload = {
-                            "type": "content_block_delta",
-                            "index": block_index,
-                            "delta": {
-                                "type": "input_json_delta",
-                                "partial_json": args
-                            }
-                        }
-
+                        blk = tool_block_idx[idx]
                         yield (
-                            f"{_STREAM_BLOCK_DELTA}"
-                            f"{json.dumps(delta_payload)}"
-                            "\n\n"
+                            _BLK_DELTA
+                            + _dumps({
+                                "type":  "content_block_delta",
+                                "index": blk,
+                                "delta": {"type": "input_json_delta",
+                                          "partial_json": args},
+                            })
+                            + "\n\n"
                         )
 
-    except Exception as error:
-
-        logger.warning(
-            "stream error: %s",
-            error
-        )
-
-    # ---------------------------------------------------------
-    # CLOSE BLOCKS
-    # ---------------------------------------------------------
+    except Exception as exc:
+        logger.warning("stream error: %s", exc)
 
-    # Close reasoning block if it wasn't closed yet
-    if (
-        reasoning_block_idx is not None
-        and not reasoning_closed
-    ):
-
-        close_payload = {
-            "type": "content_block_stop",
-            "index": reasoning_block_idx
-        }
+    # ── CLOSE ALL OPEN BLOCKS ─────────────────────────────────────────────
 
+    if reasoning_block_idx is not None and not reasoning_closed:
+        sig = _synthetic_signature(reasoning_text_acc)
         yield (
-            f"{_STREAM_BLOCK_STOP}"
-            f"{json.dumps(close_payload)}"
-            "\n\n"
+            _BLK_DELTA
+            + _dumps({
+                "type":  "content_block_delta",
+                "index": reasoning_block_idx,
+                "delta": {"type": "signature_delta", "signature": sig},
+            })
+            + "\n\n"
+        )
+        yield (
+            _BLK_STOP
+            + _dumps({"type": "content_block_stop", "index": reasoning_block_idx})
+            + "\n\n"
         )
-
-        reasoning_closed = True
 
     if text_block_idx is not None:
-
-        close_payload = {
-            "type": "content_block_stop",
-            "index": text_block_idx
-        }
-
         yield (
-            f"{_STREAM_BLOCK_STOP}"
-            f"{json.dumps(close_payload)}"
-            "\n\n"
+            _BLK_STOP
+            + _dumps({"type": "content_block_stop", "index": text_block_idx})
+            + "\n\n"
         )
 
-    for _, block_index in tool_block_idx.items():
-
-        close_payload = {
-            "type": "content_block_stop",
-            "index": block_index
-        }
-
+    for blk in tool_block_idx.values():
         yield (
-            f"{_STREAM_BLOCK_STOP}"
-            f"{json.dumps(close_payload)}"
-            "\n\n"
+            _BLK_STOP
+            + _dumps({"type": "content_block_stop", "index": blk})
+            + "\n\n"
         )
 
-    delta_payload = {
-        "type": "message_delta",
-        "delta": {
-            "stop_reason": stop_reason,
-            "stop_sequence": None
-        },
-        "usage": {
-            "output_tokens": output_tokens
-        }
-    }
-
-    yield (
-        f"{_STREAM_MSG_DELTA}"
-        f"{json.dumps(delta_payload)}"
-        "\n\n"
-    )
-
-    stop_payload = {
-        "type": "message_stop"
-    }
-
     yield (
-        f"{_STREAM_MSG_STOP}"
-        f"{json.dumps(stop_payload)}"
-        "\n\n"
+        _MSG_DELTA
+        + _dumps({
+            "type":  "message_delta",
+            "delta": {"stop_reason": stop_reason, "stop_sequence": None},
+            "usage": {"output_tokens": output_tokens},
+        })
+        + "\n\n"
     )
+    yield f"{_MSG_STOP}" + '{"type":"message_stop"}' + "\n\n"
 
 # =========================================================
 # MAIN ENDPOINT
@@ -1002,241 +770,168 @@ async def convert_stream_to_anthropic(
 @app.post("/v1/messages")
 @app.post("/v1/chat/completions")
 async def messages(
-    request: Request,
-    authorization: str = Header(default=None)
+    request:       Request,
+    authorization: str = Header(default=None),
 ):
-
     request_id = uuid.uuid4().hex[:8]
 
     if not OPENCODE_API_KEY:
-
         return JSONResponse(
             status_code=401,
-            content={
-                "error": "Missing OPENCODE_API_KEY"
-            }
+            content={"error": "Missing OPENCODE_API_KEY"},
         )
 
     try:
-
         body = await request.json()
-
     except Exception:
+        return JSONResponse(status_code=400, content={"error": "Invalid JSON"})
 
-        return JSONResponse(
-            status_code=400,
-            content={
-                "error": "Invalid JSON"
-            }
-        )
-
-    requested_model = body.get(
-        "model",
-        "claude-sonnet-4-6"
-    )
-
-    mapped_model = MODEL_MAP.get(
-        requested_model,
-        DEFAULT_MODEL
-    )
-
-    stream = bool(
-        body.get("stream", False)
-    )
-
-    # ---------------------------------------------------------
-    # TOOLS
-    # ---------------------------------------------------------
-
-    anthropic_tools = body.get("tools", [])
-
-    openai_tools = convert_tools_to_openai(
-        anthropic_tools
-    )
+    requested_model = body.get("model", "claude-sonnet-4-6")
+    mapped_model    = MODEL_MAP.get(requested_model, DEFAULT_MODEL)
+    stream          = bool(body.get("stream", False))
+    thinking_on     = THINKING_MODE == "enabled"
 
+    anthropic_tools    = body.get("tools", [])
+    openai_tools       = convert_tools_to_openai(anthropic_tools)
     openai_tool_choice = map_tool_choice(
-        body.get("tool_choice"),
-        bool(openai_tools)
+        body.get("tool_choice"), bool(openai_tools), thinking_on
     )
 
-    # ---------------------------------------------------------
-    # MESSAGES
-    # ---------------------------------------------------------
-
-    openai_messages = normalize_messages(
-        body.get("messages", [])
-    )
+    openai_messages = normalize_messages(body.get("messages", []))
 
     system_prompt = body.get("system", "")
-
     if system_prompt:
-
-        openai_messages.insert(
-            0,
-            {
-                "role": "system",
-                "content": system_prompt
-            }
-        )
-
-    # ---------------------------------------------------------
-    # PAYLOAD
-    # ---------------------------------------------------------
+        if isinstance(system_prompt, list):
+            system_prompt = "\n".join(
+                b.get("text", "") if isinstance(b, dict) else str(b)
+                for b in system_prompt
+            )
+        openai_messages.insert(0, {"role": "system", "content": system_prompt})
 
     payload = build_payload(
-        body,
-        openai_messages,
-        openai_tools,
-        openai_tool_choice,
-        mapped_model,
-        stream
+        body, openai_messages, openai_tools,
+        openai_tool_choice, mapped_model, stream,
     )
-
-    headers = {
-        **UPSTREAM_HEADERS,
-        "x-api-key": OPENCODE_API_KEY
-    }
+    headers = _make_upstream_headers(OPENCODE_API_KEY)
 
     logger.info(
-        "[%s] %s -> %s | stream=%s | tools=%d | "
-        "thinking=%s | msgs=%d",
-        request_id,
-        requested_model,
-        mapped_model,
-        stream,
-        len(openai_tools),
-        THINKING_MODE,
-        len(openai_messages),
+        "[%s] %s→%s stream=%s tools=%d thinking=%s msgs=%d",
+        request_id, requested_model, mapped_model,
+        stream, len(openai_tools), THINKING_MODE, len(openai_messages),
     )
 
-    client = request.app.state.client
-
+    client     = request.app.state.client
     last_error = None
 
-    # =========================================================
-    # RETRIES
-    # =========================================================
-
     for attempt in range(1, MAX_RETRIES + 1):
+        # only delay on retry, not on first attempt
+        if attempt > 1:
+            await asyncio.sleep(0.5 * (attempt - 1))
 
         try:
-
-            # -------------------------------------------------
-            # STREAM MODE
-            # -------------------------------------------------
-
+            # ── STREAM MODE ───────────────────────────────────────────────
             if stream:
-
-                upstream_request = client.build_request(
-                    "POST",
-                    UPSTREAM_URL,
-                    json=payload,
-                    headers=headers
+                upstream_req = client.build_request(
+                    "POST", UPSTREAM_URL, json=payload, headers=headers
                 )
+                response = await client.send(upstream_req, stream=True)
 
-                response = await client.send(
-                    upstream_request,
-                    stream=True
-                )
-
-                if response.status_code != 200:
-
-                    raw_error = (
-                        await response.aread()
-                    ).decode(errors="ignore")
-
+                if response.status_code == 429:
+                    retry_after = float(response.headers.get("retry-after", "0"))
+                    raw_err = (await response.aread()).decode(errors="ignore")
                     logger.warning(
-                        "[%s] fail %d: %s",
-                        request_id,
-                        response.status_code,
-                        raw_error[:300]
+                        "[%s] 429 — retry-after=%.0fs — body: %s",
+                        request_id, retry_after, raw_err[:400]
+                    )
+                    await response.aclose()
+                    # Do NOT sleep retry-after when it's large (e.g. 7000s = 2h).
+                    # The quota is exhausted on the key — sleeping won't help.
+                    # Return a clear 429 to the client immediately.
+                    return JSONResponse(
+                        status_code=429,
+                        content={
+                            "type": "error",
+                            "error": {
+                                "type": "rate_limit_error",
+                                "message": (
+                                    f"Upstream rate-limited. "
+                                    f"retry-after={retry_after:.0f}s. "
+                                    f"Detail: {raw_err[:200]}"
+                                ),
+                            },
+                        },
+                        headers={"retry-after": str(int(retry_after))} if retry_after else {},
                     )
 
-                    last_error = raw_error
-
-                    await asyncio.sleep(RETRY_DELAY)
-
+                if response.status_code != 200:
+                    raw_err = (await response.aread()).decode(errors="ignore")
+                    logger.warning("[%s] upstream %d: %s",
+                                   request_id, response.status_code, raw_err[:300])
+                    last_error = raw_err
+                    await response.aclose()
                     continue
 
+                # BackgroundTask closes the connection after the generator exhausts
                 return StreamingResponse(
-                    convert_stream_to_anthropic(
-                        response,
-                        requested_model
-                    ),
+                    convert_stream_to_anthropic(response, requested_model),
                     media_type="text/event-stream",
                     headers=STREAM_RESPONSE_HEADERS,
+                    background=BackgroundTask(response.aclose),
                 )
 
-            # -------------------------------------------------
-            # NORMAL MODE
-            # -------------------------------------------------
-
+            # ── NORMAL MODE ───────────────────────────────────────────────
             response = await client.post(
-                UPSTREAM_URL,
-                json=payload,
-                headers=headers
+                UPSTREAM_URL, json=payload, headers=headers
             )
 
-            if response.status_code != 200:
-
+            if response.status_code == 429:
+                retry_after = float(response.headers.get("retry-after", "0"))
                 logger.warning(
-                    "[%s] fail %d",
-                    request_id,
-                    response.status_code
+                    "[%s] 429 — retry-after=%.0fs — body: %s",
+                    request_id, retry_after, response.text[:400]
+                )
+                return JSONResponse(
+                    status_code=429,
+                    content={
+                        "type": "error",
+                        "error": {
+                            "type": "rate_limit_error",
+                            "message": (
+                                f"Upstream rate-limited. "
+                                f"retry-after={retry_after:.0f}s. "
+                                f"Detail: {response.text[:200]}"
+                            ),
+                        },
+                    },
+                    headers={"retry-after": str(int(retry_after))} if retry_after else {},
                 )
 
+            if response.status_code != 200:
+                logger.warning("[%s] upstream %d", request_id, response.status_code)
                 last_error = response.text
-
-                await asyncio.sleep(RETRY_DELAY)
-
                 continue
 
-            raw_response = response.text.strip()
-
-            if not raw_response:
-
+            raw = response.content
+            if not raw or not raw.strip():
                 return JSONResponse(
-                    status_code=502,
-                    content={
-                        "error": "Empty upstream response"
-                    }
+                    status_code=502, content={"error": "Empty upstream response"}
                 )
-
             try:
-
-                parsed_response = response.json()
-
+                parsed = _loads(raw)
             except Exception:
-
                 return JSONResponse(
-                    status_code=502,
-                    content={
-                        "error": "Invalid upstream JSON"
-                    }
+                    status_code=502, content={"error": "Invalid upstream JSON"}
                 )
 
             return JSONResponse(
-                content=convert_to_anthropic(
-                    parsed_response,
-                    requested_model
-                )
+                content=convert_to_anthropic(parsed, requested_model)
             )
 
-        except Exception as error:
-
-            last_error = str(error)
-
-            logger.exception(
-                "[%s] exception",
-                request_id
-            )
-
-            await asyncio.sleep(RETRY_DELAY)
+        except Exception as exc:
+            last_error = str(exc)
+            logger.exception("[%s] exception attempt %d", request_id, attempt)
 
     return JSONResponse(
         status_code=502,
-        content={
-            "error": "Upstream failed",
-            "details": last_error
-        }
-    )
+        content={"error": "Upstream failed", "details": last_error},
+    )