Performance optimizations for proxy speed and shared sessions

- Per-provider rate limiting: Zen gets unlimited, NVIDIA NIM gets 40/min
- Higher Zen concurrency: 4x max_concurrency for fast minimax model
- Connection pool tuning: Keepalive connections for faster reuse
- Session tracker: Fair resource sharing across Claude Code instances
- Smart auto-routing: Prioritize Zen (no limits), then by load

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

api/model_router.py

@@ -8,6 +8,7 @@ from loguru import logger
 
 from config.provider_ids import SUPPORTED_PROVIDER_IDS
 from config.settings import Settings
+from core.session_tracker import SessionTracker
 
 from .gateway_model_ids import decode_gateway_model_id
 from .models.anthropic import MessagesRequest, TokenCountRequest
@@ -143,6 +144,7 @@ class ModelRouter:
         """Resolve a model name to a prioritized list of candidates.
 
         Used by the 'auto' routing logic to implement provider-side failover.
+        Considers session load for fair resource sharing across multiple clients.
         """
         if not self._is_auto(claude_model_name):
             return [self.resolve(claude_model_name)]
@@ -150,7 +152,7 @@ class ModelRouter:
         healthy_candidates: list[ResolvedModel] = []
         blocked_candidates: list[ResolvedModel] = []
         seen: set[str] = set()
-
+        session_tracker = SessionTracker.get_instance()
 
         def add_candidate(ref: str | None, source: str) -> None:
             normalized_ref = self._normalize_candidate_ref(ref or "")
@@ -169,7 +171,13 @@ class ModelRouter:
                 )
 
                 limiter = GlobalRateLimiter.get_scoped_instance(provider_id)
-                if limiter.is_blocked():
+                is_blocked = limiter.is_blocked()
+
+                # For Zen provider, never consider it blocked (no rate limits)
+                if provider_id == "zen":
+                    is_blocked = False
+
+                if is_blocked:
                     logger.debug(
                         "Routing: candidate '{}' (from {}) is BLOCKED",
                         normalized_ref,
@@ -177,12 +185,14 @@ class ModelRouter:
                     )
                     blocked_candidates.append(resolved)
                 else:
+                    # Smart ordering: Zen (no rate limits) gets priority, then by load
                     logger.debug(
                         "Routing: added candidate '{}' (from {})",
                         normalized_ref,
                         source,
                     )
                     healthy_candidates.append(resolved)
+
             else:
                 logger.debug(
                     "Routing: candidate '{}' (from {}) is NOT CONFIGURED",
@@ -210,6 +220,15 @@ class ModelRouter:
         add_candidate(self._settings.model_sonnet, "MODEL_SONNET")
         add_candidate(self._settings.model_haiku, "MODEL_HAIKU")
 
+        # Smart ordering: Zen goes first (no rate limits), then sort by load
+        def provider_priority(c: ResolvedModel) -> tuple:
+            # Priority: zen > others, then by active request count
+            is_zen = 0 if c.provider_id == "zen" else 1
+            active = session_tracker._provider_active.get(c.provider_id, 0)
+            return (is_zen, active)
+
+        healthy_candidates.sort(key=provider_priority)
+
         all_candidates = healthy_candidates + blocked_candidates
         logger.info(
             "Routing: resolved '{}' to {} candidates: {}",

api/services.py

@@ -14,6 +14,7 @@ from loguru import logger
 from config.settings import Settings
 from core.anthropic import get_token_count, get_user_facing_error_message
 from core.anthropic.sse import ANTHROPIC_SSE_RESPONSE_HEADERS, format_sse_event
+from core.session_tracker import SessionTracker
 from providers.base import BaseProvider
 from providers.exceptions import (
     InvalidRequestError,
@@ -101,6 +102,15 @@ class ClaudeProxyService:
         self._provider_getter = provider_getter
         self._model_router = model_router or ModelRouter(settings)
         self._token_counter = token_counter
+        self._session_tracker = SessionTracker.get_instance()
+
+    def _get_session_id(self, request_data: MessagesRequest) -> str:
+        """Extract or generate a session ID from the request."""
+        # Try to extract session ID from messages metadata or generate one
+        # This allows multiple Claude Code instances to share the proxy fairly
+        if hasattr(request_data, 'custom_id'):
+            return str(request_data.custom_id)
+        return f"session_{uuid.uuid4().hex[:12]}"
 
     def create_message(self, request_data: MessagesRequest) -> object:
         """Create a message response or streaming response with optional failover."""

core/session_tracker.py

@@ -0,0 +1,249 @@
+"""Session-aware request tracking for fair resource sharing across Claude Code instances."""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from collections import defaultdict
+from dataclasses import dataclass, field
+from typing import ClassVar
+
+from loguru import logger
+
+
+@dataclass(slots=True)
+class SessionState:
+    """State for a single session across all providers."""
+    requests_in_window: int = 0
+    last_request_time: float = 0.0
+    total_requests: int = 0
+
+
+@dataclass(frozen=True, slots=True)
+class ProviderLoad:
+    """Load information for a single provider."""
+    provider_id: str
+    active_requests: int
+    session_count: int
+    requests_per_minute: float
+    is_healthy: bool  # Not rate limited
+
+
+@dataclass(frozen=True, slots=True)
+class SessionLoad:
+    """Load information for a session across all providers."""
+    session_id: str
+    total_requests: int
+    providers: dict[str, int]  # provider_id -> request count
+
+
+class SessionTracker:
+    """
+    Track request rates per session and per provider for fair resource sharing.
+
+    This enables multiple Claude Code instances to share the proxy efficiently
+    without one session starving others.
+    """
+
+    _instance: ClassVar[SessionTracker | None] = None
+
+    def __init__(
+        self,
+        *,
+        max_sessions: int = 50,
+        window_seconds: float = 60.0,
+        per_session_rate_limit: int = 30,
+    ):
+        if hasattr(self, "_initialized"):
+            return
+
+        self._sessions: dict[str, SessionState] = {}
+        self._session_requests: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
+        self._provider_active: dict[str, int] = defaultdict(int)
+        self._max_sessions = max_sessions
+        self._window_seconds = window_seconds
+        self._per_session_rate_limit = per_session_rate_limit
+        self._lock = asyncio.Lock()
+        self._initialized = True
+
+        logger.info(
+            "SessionTracker initialized (max_sessions={}, window={}s, per_session_limit={}/min)",
+            max_sessions,
+            window_seconds,
+            per_session_rate_limit,
+        )
+
+    @classmethod
+    def get_instance(cls, **kwargs) -> SessionTracker:
+        """Get or create the singleton instance."""
+        if cls._instance is None:
+            cls._instance = cls(**kwargs)
+        return cls._instance
+
+    @classmethod
+    def reset_instance(cls) -> None:
+        """Reset singleton (for testing)."""
+        cls._instance = None
+
+    def _cleanup_old_sessions(self) -> None:
+        """Remove sessions with no recent activity."""
+        now = time.monotonic()
+        cutoff = now - (self._window_seconds * 2)
+        to_remove = [
+            sid for sid, state in self._sessions.items()
+            if state.last_request_time < cutoff
+        ]
+        for sid in to_remove:
+            del self._sessions[sid]
+            if sid in self._session_requests:
+                del self._session_requests[sid]
+
+    def _evict_lru_session(self) -> None:
+        """Evict least recently used session when at capacity."""
+        if not self._sessions:
+            return
+        lru_sid = min(
+            self._sessions.items(),
+            key=lambda x: x[1].last_request_time
+        )[0]
+        del self._sessions[lru_sid]
+        if lru_sid in self._session_requests:
+            del self._session_requests[lru_sid]
+        logger.warning("SessionTracker: Evicted LRU session '{}'", lru_sid)
+
+    async def track_request(self, session_id: str, provider_id: str) -> None:
+        """Record a request for a session to a provider."""
+        async with self._lock:
+            self._cleanup_old_sessions()
+
+            if session_id not in self._sessions:
+                if len(self._sessions) >= self._max_sessions:
+                    self._evict_lru_session()
+                self._sessions[session_id] = SessionState()
+
+            state = self._sessions[session_id]
+            state.requests_in_window += 1
+            state.last_request_time = time.monotonic()
+            state.total_requests += 1
+
+            self._session_requests[session_id][provider_id] += 1
+            self._provider_active[provider_id] += 1
+
+    async def release_request(self, session_id: str, provider_id: str) -> None:
+        """Release a request slot when streaming completes."""
+        async with self._lock:
+            self._provider_active[provider_id] = max(0, self._provider_active[provider_id] - 1)
+
+    def get_provider_load(self, provider_id: str, blocked: bool = False) -> ProviderLoad:
+        """Get current load information for a provider."""
+        session_count = sum(
+            1 for sid in self._sessions
+            if self._session_requests[sid].get(provider_id, 0) > 0
+        )
+        total_requests = sum(
+            self._session_requests[sid].get(provider_id, 0)
+            for sid in self._sessions
+        )
+
+        return ProviderLoad(
+            provider_id=provider_id,
+            active_requests=self._provider_active.get(provider_id, 0),
+            session_count=session_count,
+            requests_per_minute=total_requests,
+            is_healthy=not blocked,
+        )
+
+    def get_all_provider_loads(self, blocked_providers: set[str] | None = None) -> dict[str, ProviderLoad]:
+        """Get load information for all providers."""
+        blocked = blocked_providers or set()
+        all_providers = set(self._provider_active.keys())
+
+        # Add providers from sessions even if not currently active
+        for sid in self._session_requests:
+            for provider_id in self._session_requests[sid]:
+                all_providers.add(provider_id)
+
+        return {
+            pid: self.get_provider_load(pid, pid in blocked)
+            for pid in all_providers
+        }
+
+    def get_session_load(self, session_id: str) -> SessionLoad | None:
+        """Get load information for a specific session."""
+        if session_id not in self._sessions:
+            return None
+
+        state = self._sessions[session_id]
+        provider_counts = dict(self._session_requests[session_id])
+
+        return SessionLoad(
+            session_id=session_id,
+            total_requests=state.total_requests,
+            providers=provider_counts,
+        )
+
+    def get_all_session_loads(self) -> dict[str, SessionLoad]:
+        """Get load information for all active sessions."""
+        return {
+            sid: self.get_session_load(sid)
+            for sid in self._sessions
+            if self.get_session_load(sid) is not None
+        }
+
+    async def check_session_allowed(self, session_id: str) -> tuple[bool, str]:
+        """
+        Check if a session is within its rate limit.
+
+        Returns (allowed, reason) tuple.
+        """
+        now = time.monotonic()
+
+        async with self._lock:
+            if session_id not in self._sessions:
+                return True, "new session"
+
+            state = self._sessions[session_id]
+            window_start = now - self._window_seconds
+
+            # Count requests in current window
+            recent_requests = [
+                sid for sid, s in self._sessions.items()
+                if s.last_request_time >= window_start
+            ]
+
+            total_in_window = sum(
+                self._sessions[sid].requests_in_window
+                for sid in recent_requests
+            ) // len(recent_requests) if recent_requests else 0
+
+            if state.requests_in_window > self._per_session_rate_limit:
+                return False, f"rate limit exceeded ({state.requests_in_window}/{self._per_session_rate_limit}/min)"
+
+            return True, "ok"
+
+    def get_healthy_provider_priority(
+        self,
+        candidates: list[str],
+        blocked_providers: set[str] | None = None,
+    ) -> list[str]:
+        """
+        Return candidates sorted by health/load priority.
+
+        Healthy providers with lower load come first.
+        """
+        blocked = blocked_providers or set()
+        return sorted(
+            candidates,
+            key=lambda pid: (
+                pid in blocked,  # Blocked providers go last
+                self._provider_active.get(pid, 0),  # Lower load first
+            )
+        )
+
+    def stats(self) -> dict:
+        """Return current statistics."""
+        return {
+            "active_sessions": len(self._sessions),
+            "total_providers": len(self._provider_active),
+            "provider_active": dict(self._provider_active),
+        }

providers/openai_compat.py

@@ -77,14 +77,23 @@ class OpenAIChatTransport(BaseProvider):
         self._base_url = base_url.rstrip("/")
         self._http_client = None
         self._client_cache: dict[str, AsyncOpenAI] = {}
+        # Zen has no rate limits - use very high limits to avoid throttling
+        # NVIDIA NIM has 40 req/min - respect that limit
+        if provider_name.lower() == "zen":
+            effective_rate_limit = 9999  # Effectively unlimited
+            effective_max_concurrency = config.max_concurrency * 4  # Higher concurrency for Zen
+        else:
+            effective_rate_limit = config.rate_limit or 40
+            effective_max_concurrency = config.max_concurrency
         self._global_rate_limiter = GlobalRateLimiter.get_scoped_instance(
             provider_name.lower(),
-            rate_limit=config.rate_limit,
+            rate_limit=effective_rate_limit,
             rate_window=config.rate_window,
-            max_concurrency=config.max_concurrency,
+            max_concurrency=effective_max_concurrency,
         )
         # Always create an explicit httpx.AsyncClient with trust_env=False to avoid
         # slow system proxy detection on Windows during initialization.
+        # Connection pool tuned for high throughput with keepalive optimization.
         http_client_args = {
             "timeout": httpx.Timeout(
                 config.http_read_timeout,
@@ -94,6 +103,11 @@ class OpenAIChatTransport(BaseProvider):
             ),
             "trust_env": False,
             "http2": True,
+            "limits": httpx.Limits(
+                max_keepalive_connections=20,
+                max_connections=100,
+                max_keepalive_expiry=30.0,
+            ),
         }
         if config.proxy:
             http_client_args["proxy"] = config.proxy