fix: hold FlowContext for full stream lifetime via finally block to prevent WARP rotation mid-stream; feat: add Anthropic /v1/messages endpoint; docs: update README with Anthropic, custom proxy, and corrected opencode.jsonc config

Author: alztrk

README.md

@@ -15,11 +15,14 @@ A microservice-architected Cloudflare WARP IP rotator and proxy server for OpenC
 
 - **Guaranteed IP Diversity**: Uses `warp-cli registration delete & new` cycles to request a distinct public IP address on every rotation.
 - **Microservices Architecture**: Decoupled `proxy-server` (FastAPI) and `warp-rotator` (Cloudflare WARP daemon) services built with Docker Compose.
-- **Clean Management Dashboard**: Lightweight Web UI built with Tailwind CSS displaying active connections, current location, token statistics, and manual rotation controls.
+- **Clean Management Dashboard**: Lightweight Web UI displaying active connections, current location, token statistics, and manual rotation controls.
 - **SQLite Data Persistence**: Stores token consumption, model request counts, and historical IP rotation logs on disk.
 - **USD Savings Calculator**: Estimates cost savings per model based on prompt and completion token rates.
 - **Table Pagination**: Built-in 5-item pagination for model usage and IP rotation log tables.
-- **Active Flow Locking**: Protects active Server-Sent Events (SSE) streams from being interrupted during an IP rotation.
+- **Active Flow Locking**: Protects active SSE streams from being interrupted during IP rotation — `_active_flows_count` is held for the **full lifetime of the generator**, not just until `return`.
+- **Smart Rate Limit Detection**: Distinguishes IP-level blocks (triggers WARP rotation) from model-level quota limits (skips rotation to prevent TCP socket teardown).
+- **Anthropic API Compatibility**: Native `/v1/messages` endpoint for Claude clients and the Vercel AI SDK `@ai-sdk/anthropic` provider.
+- **Custom Proxy Pool Support**: Round-robin outbound proxy pool via `data/proxies.txt` or `PROXY_LIST` environment variable.
 
 ---
 
@@ -120,14 +123,18 @@ python manager.py
 
 ## Configuration
 
-To use the local proxy server within OpenCode, update your configuration file located at `~/.config/opencode/opencode.jsonc`:
+### OpenAI-Compatible Provider (Default)
+
+To use the local proxy server within OpenCode, update your configuration file at `~/.config/opencode/opencode.jsonc`:
 
 ```jsonc
 {
   "provider": {
     "opencode-zen-local": {
+      "npm": "@ai-sdk/openai-compatible",
       "options": {
-        "baseURL": "http://127.0.0.1:8000/v1"
+        "baseURL": "http://127.0.0.1:8000/v1",
+        "apiKey": "any"
       },
       "name": "OpenCode Zen Local Proxy"
     }
@@ -139,11 +146,53 @@ To use the local proxy server within OpenCode, update your configuration file lo
 
 ---
 
+### Anthropic API Provider
+
+The proxy exposes a native Anthropic-compatible `/v1/messages` endpoint. To use it with OpenCode's Anthropic provider:
+
+```jsonc
+{
+  "provider": {
+    "my-anthropic-proxy": {
+      "npm": "@ai-sdk/anthropic",
+      "options": {
+        "baseURL": "http://127.0.0.1:8000",
+        "apiKey": "any"
+      }
+    }
+  }
+}
+```
+
+> Requests sent to `/v1/messages` are translated to OpenAI format internally and routed through the same WARP-protected upstream.
+
+---
+
+### Custom Outbound Proxy Pool
+
+If you want to use your own HTTP/SOCKS5 proxies instead of (or in addition to) Cloudflare WARP:
+
+**Option 1 — File:** Create `data/proxies.txt` with one proxy per line:
+```
+http://user:pass@proxy1.example.com:8080
+socks5://proxy2.example.com:1080
+```
+
+**Option 2 — Environment variable:**
+```bash
+PROXY_LIST="http://proxy1:8080,socks5://proxy2:1080" docker compose up -d
+```
+
+The proxy pool rotates in round-robin order across all outbound requests.
+
+---
+
 ## API Endpoints Reference
 
 | Endpoint | Method | Description |
 | :--- | :--- | :--- |
 | `/v1/chat/completions` | `POST` | OpenAI-compatible chat completion endpoint with automatic retry and IP rotation. |
+| `/v1/messages` | `POST` | Anthropic-compatible endpoint (`/v1/messages`) for Claude clients and `@ai-sdk/anthropic`. |
 | `/v1/models` | `GET` | Returns list of currently discovered active free models. |
 | `/dashboard` | `GET` | Renders the HTML Web Management Dashboard. |
 | `/metrics` | `GET` | Returns structured JSON metrics including verified IP, uptime, and request counters. |

server.py

@@ -675,30 +675,45 @@ def get_realistic_headers() -> Dict[str, str]:
     }
 
 async def stream_openai_response(response, model_name: str) -> AsyncGenerator[bytes, None]:
+    """
+    Raw byte passthrough SSE stream generator.
+    CRITICAL: FlowContext is held for the ENTIRE duration of streaming so that
+    rotate_warp() in rotator.py sees _active_flows_count > 0 and skips IP rotation
+    while a response is still being streamed to the client.
+    """
     loop = asyncio.get_event_loop()
-    with FlowContext():
-        try:
-            def get_next_chunk(iter_content):
-                try:
-                    return next(iter_content)
-                except StopIteration:
-                    return None
-
-            # Stream raw bytes chunks directly from HTTP response (Do not split lines or strip whitespace/newlines)
-            chunk_iter = response.iter_content(chunk_size=4096)
-            
-            while True:
-                chunk = await loop.run_in_executor(None, get_next_chunk, chunk_iter)
-                if chunk is None or not chunk:
-                    break
-                
-                # Raw chunk passthrough to client
-                yield chunk
+    global _active_flows_count
 
-            yield b"\ndata: [DONE]\n\n"
-        except Exception as e:
-            log.error(f"Stream exception caught: {e}")
-            yield b"\ndata: [DONE]\n\n"
+    # Manually increment flow counter — keeps it raised until generator is exhausted/closed
+    with _flow_lock:
+        _active_flows_count += 1
+
+    try:
+        def get_next_chunk(iter_content):
+            try:
+                return next(iter_content)
+            except StopIteration:
+                return None
+
+        chunk_iter = response.iter_content(chunk_size=4096)
+
+        while True:
+            chunk = await loop.run_in_executor(None, get_next_chunk, chunk_iter)
+            if chunk is None or not chunk:
+                break
+            yield chunk
+
+        yield b"\ndata: [DONE]\n\n"
+    except GeneratorExit:
+        # Client disconnected — still need to decrement
+        pass
+    except Exception as e:
+        log.error(f"Stream exception caught: {e}")
+        yield b"\ndata: [DONE]\n\n"
+    finally:
+        # ALWAYS decrement when stream ends, regardless of how it ended
+        with _flow_lock:
+            _active_flows_count = max(0, _active_flows_count - 1)
 
 @app.get("/dashboard", response_class=HTMLResponse)
 async def dashboard():
@@ -787,59 +802,66 @@ async def chat_completions(raw_request: Request):
 
     for attempt in range(1, MAX_RETRIES_ON_429 + 1):
         try:
-            with FlowContext():
-                from curl_cffi import requests as cffi_requests
-                import random
-                
-                time.sleep(random.uniform(0.1, 0.3))
-                proxies = get_next_outbound_proxy()
-
-                response = cffi_requests.post(
-                    TARGET_ZEN_URL,
-                    json=payload,
-                    headers=headers,
-                    impersonate="chrome124",
-                    stream=is_stream,
-                    proxies=proxies,
-                    timeout=120
-                )
-                
-                # Check HTTP status code for rate-limit or upstream error
-                if response.status_code == 429 or response.status_code >= 500:
-                    metrics["rate_limited_requests"] += 1
-                    err_text = response.text.lower()
-                    
-                    # Distinguish Model/Provider-level limits vs. IP-level blocks
-                    is_model_specific_limit = any(k in err_text for k in ["model_rate_limit", "quota_exceeded", "per_model_limit", "credit_balance"])
-                    
-                    import random
-                    delay = (INITIAL_BACKOFF * (2 ** (attempt - 1))) + random.uniform(0.5, 1.5)
-                    
-                    if is_model_specific_limit:
-                        log.warning(f"Model-level limit for '{current_model}'. Skipping IP rotation to prevent socket teardown. Retrying in {delay:.2f}s...")
-                        time.sleep(delay)
-                        continue
-                    else:
-                        log.warning(f"HTTP {response.status_code} (IP/Network block) for '{current_model}'. Triggering IP Rotation & Retrying in {delay:.2f}s...")
-                        rotate_warp(reason=f"HTTP {response.status_code} on {current_model}")
-                        time.sleep(delay)
-                        continue
-
-                metrics["successful_requests"] += 1
-                
-                if is_stream:
-                    track_token_usage(current_model, prompt_tokens=100, completion_tokens=150)
-                    return StreamingResponse(
-                        stream_openai_response(response, current_model),
-                        media_type="text/event-stream"
-                    )
+            from curl_cffi import requests as cffi_requests
+            import random
+
+            time.sleep(random.uniform(0.1, 0.3))
+            proxies = get_next_outbound_proxy()
+
+            # FlowContext is NOT used here for streaming — the generator manages it internally
+            # For non-streaming we use it to block rotation during the entire request
+            response = cffi_requests.post(
+                TARGET_ZEN_URL,
+                json=payload,
+                headers=headers,
+                impersonate="chrome124",
+                stream=is_stream,
+                proxies=proxies,
+                timeout=120
+            )
+
+            # Check HTTP status code for rate-limit or upstream error
+            if response.status_code == 429 or response.status_code >= 500:
+                metrics["rate_limited_requests"] += 1
+                err_text = response.text.lower()
+
+                # Distinguish Model/Provider-level limits vs. IP-level blocks
+                is_model_specific_limit = any(k in err_text for k in ["model_rate_limit", "quota_exceeded", "per_model_limit", "credit_balance", "insufficient_quota"])
+
+                delay = (INITIAL_BACKOFF * (2 ** (attempt - 1))) + random.uniform(0.5, 1.5)
+
+                if is_model_specific_limit:
+                    log.warning(f"Model-level limit for '{current_model}'. Skipping IP rotation to prevent socket teardown. Retrying in {delay:.2f}s...")
+                    time.sleep(delay)
+                    continue
                 else:
+                    log.warning(f"HTTP {response.status_code} (IP/Network block) for '{current_model}'. Triggering IP Rotation & Retrying in {delay:.2f}s...")
+                    rotate_warp(reason=f"HTTP {response.status_code} on {current_model}")
+                    time.sleep(delay)
+                    continue
+
+            metrics["successful_requests"] += 1
+
+            if is_stream:
+                track_token_usage(current_model, prompt_tokens=100, completion_tokens=150)
+                # stream_openai_response manages FlowContext internally via finally block
+                return StreamingResponse(
+                    stream_openai_response(response, current_model),
+                    media_type="text/event-stream",
+                    headers={
+                        "Cache-Control": "no-cache",
+                        "X-Accel-Buffering": "no",
+                        "Connection": "keep-alive",
+                    }
+                )
+            else:
+                with FlowContext():
                     try:
                         res_json = response.json()
                         usage = res_json.get("usage", {})
                         track_token_usage(
-                            current_model, 
-                            prompt_tokens=usage.get("prompt_tokens", 50), 
+                            current_model,
+                            prompt_tokens=usage.get("prompt_tokens", 50),
                             completion_tokens=usage.get("completion_tokens", 100)
                         )
                         return JSONResponse(content=res_json)