reraising is working now

xzrderek · xzrderek · commit c705cb8d88a8 · 2025-10-29T04:00:08.000-07:00
diff --git a/eval_protocol/models.py b/eval_protocol/models.py
@@ -214,10 +214,10 @@ def raise_from_status_details(cls, status_details: List[Dict[str, Any]]) -> bool
                         logger.info(f"Re-raising {exception_type} from status details")
                         raise exception_to_raise
                     else:
-                        logger.debug(f"Could not create instance of {exception_type}")
+                        logger.info(f"Could not create instance of {exception_type}")
                         continue
                 else:
-                    logger.debug(f"Could not import exception type: {exception_type}")
+                    logger.info(f"Could not import exception type: {exception_type}")
                     continue
 
         return False
@@ -244,7 +244,9 @@ def _create_exception_instance(cls, exception_class: type, message: str) -> Opti
             # NOTE: we are losing some diagnostic information here by not passing the model and llm_provider. We could try to capture full exception state in rollout_error_from_exception.
             lambda: exception_class(message, model="unknown", llm_provider="unknown"),
             lambda: exception_class(message=message, model="unknown", llm_provider="unknown"),
-            # Pattern 4: No arguments (fallback)
+            # Pattern 5: OpenAI exceptions - create mock response object
+            lambda: cls._create_openai_exception(exception_class, message),
+            # Pattern 7: No arguments (fallback)
             lambda: exception_class(),
         ]
 
@@ -260,6 +262,36 @@ def _create_exception_instance(cls, exception_class: type, message: str) -> Opti
         logger.debug(f"All constructor patterns failed for {exception_class.__name__}")
         return None
 
+    @classmethod
+    def _create_openai_exception(cls, exception_class: type, message: str) -> Optional[Exception]:
+        """
+        Create OpenAI exception with a mock response object.
+
+        OpenAI exceptions require httpx.Response objects which are complex to create,
+        so we create a minimal mock that satisfies the basic requirements.
+        """
+        try:
+            import httpx
+
+            # Create a minimal mock response object
+            class MockRequest:
+                def __init__(self):
+                    self.method = "POST"
+                    self.url = "https://api.openai.com/v1/chat/completions"
+
+            class MockResponse:
+                def __init__(self):
+                    self.status_code = 404
+                    self.headers = {"x-request-id": "mock-request-id"}
+                    self.request = MockRequest()
+
+            mock_response = MockResponse()
+            return exception_class(message, response=mock_response, body=None)
+
+        except Exception as e:
+            logging.getLogger(__name__).debug(f"Failed to create OpenAI exception with mock response: {e}")
+            return None
+
     @classmethod
     def _import_exception_class(cls, exception_type: str) -> Optional[type]:
         """
diff --git a/eval_protocol/quickstart/svg_agent/vercel_svg_server/api/init.py b/eval_protocol/quickstart/svg_agent/vercel_svg_server/api/init.py
@@ -9,7 +9,8 @@
 import os
 import logging
 import sys
-from http.server import BaseHTTPRequestHandler
+import asyncio
+from flask import Flask, request, jsonify
 from openai import OpenAI
 from dotenv import load_dotenv
 
@@ -44,119 +45,133 @@ def filter(self, record: logging.LogRecord) -> bool:
 # Attach Fireworks tracing handler to root logger (non-stream HTTP sink)
 root_logger.addHandler(FireworksTracingHttpHandler())
 
+# Create Flask app
+app = Flask(__name__)
+
+
+async def execute_rollout_background(req, api_key):
+    """Execute the OpenAI completion in background and log results"""
+    # Attach rollout_id filter to logger
+    logger = logging.getLogger(f"{__name__}.{req.metadata.rollout_id}")
+    logger.addFilter(RolloutIdFilter(req.metadata.rollout_id))
+
+    try:
+        model = req.completion_params.get("model")
+        # Uncomment if you need to strip fireworks_ai/ prefix
+        # if model and isinstance(model, str) and model.startswith("fireworks_ai/"):
+        #     model = model[len("fireworks_ai/"):]
+
+        # Prepare completion arguments
+        completion_kwargs = {
+            "messages": req.messages,
+            # "messages": [{"role": "user", "content": "Hello, how are you?"}],
+            "model": model,
+            "temperature": req.completion_params.get("temperature"),
+            "max_tokens": req.completion_params.get("max_tokens"),
+        }
+
+        # Add tools if present
+        if req.tools:
+            completion_kwargs["tools"] = req.tools
+
+        logger.info(
+            f"DEBUG: {req.model_base_url}, COMPLETION_KWARGS: {completion_kwargs}, API_KEY: {api_key}, MODEL: {model}"
+        )
+
+        # Create AsyncOpenAI client
+        # client = AsyncOpenAI(base_url=req.model_base_url, api_key=api_key)
+        client = OpenAI(base_url=req.model_base_url, api_key=api_key)
+
+        logger.info(f"Sending completion request to model {model}")
 
-class handler(BaseHTTPRequestHandler):
-    def do_POST(self):
-        try:
-            # Read and parse request body
-            content_length = int(self.headers.get("Content-Length", 0))
-            request_body = self.rfile.read(content_length).decode("utf-8")
-            request_data = json.loads(request_body)
-
-            # Parse as InitRequest
-            req = InitRequest(**request_data)
-
-            # Attach rollout_id filter to logger
-            logger = logging.getLogger(f"{__name__}.{req.metadata.rollout_id}")
-            logger.addFilter(RolloutIdFilter(req.metadata.rollout_id))
-
-            # Validate required fields
-            if not req.messages:
-                error_msg = "messages is required"
-                logger.error(error_msg, extra={"status": Status.rollout_error(error_msg)})
-                self._send_error(400, error_msg)
-                return
-
-            model = req.completion_params.get("model")
-            if model and isinstance(model, str) and model.startswith("fireworks_ai/"):
-                model = model[len("fireworks_ai/") :]
-
-            # Prepare completion arguments
-            completion_kwargs = {
-                "messages": req.messages,
-                "model": model,
-                "temperature": req.completion_params.get("temperature"),
-                "max_tokens": req.completion_params.get("max_tokens"),
-            }
-
-            # Add tools if present
-            if req.tools:
-                completion_kwargs["tools"] = req.tools
-
-            # Get API key (prefer request api_key, fallback to environment)
-            api_key = req.api_key or os.environ.get("FIREWORKS_API_KEY")
-            if not api_key:
-                error_msg = "API key not provided in request or FIREWORKS_API_KEY environment variable"
-                logger.error(error_msg, extra={"status": Status.rollout_error(error_msg)})
-                self._send_error(500, error_msg)
-                return
-
-            # Create OpenAI client
-            client = OpenAI(base_url=req.model_base_url, api_key=api_key)
-
-            logger.info(f"Sending completion request to model {req.completion_params.get('model')}")
-
-            # Make the model call
-            completion = client.chat.completions.create(**completion_kwargs)
-
-            logger.info(f"Completed response: {completion}")
-
-            # Log completion status
-            logger.info(f"Rollout {req.metadata.rollout_id} completed", extra={"status": Status.rollout_finished()})
-
-            # Return the completion response
-            response_data = {
-                "status": "completed",
-                "rollout_id": req.metadata.rollout_id,
-                "choices": [
-                    {
-                        "message": {
-                            "role": completion.choices[0].message.role,
-                            "content": completion.choices[0].message.content,
-                        }
-                    }
-                ],
-            }
-
-            self._send_json_response(200, response_data)
-
-        except Exception as e:
-            # Log error if we have the request context
-            if "req" in locals() and "logger" in locals():
-                logger.error(f"❌ Error in rollout {req.metadata.rollout_id}: {e}")
-                logger.error(str(e), extra={"status": Status.rollout_error(str(e))})
-
-            self._send_error(500, str(e))
-
-    def do_GET(self):
-        """Health check endpoint"""
-        self._send_json_response(
-            200,
-            {
-                "status": "ok",
-                "message": "SVGBench Vercel Serverless Function",
-                "endpoints": {"POST /": "Process SVGBench evaluation requests"},
-            },
+        # Make the async model call with timeout
+        import time
+
+        logger.info(f"timing start: {time.time()}")
+        completion = client.chat.completions.create(**completion_kwargs)
+        logger.info(f"Completed response: {completion}")
+        logger.info(f"timing end: {time.time()}")
+        # Log successful completion - THIS IS WHAT RemoteRolloutProcessor POLLS FOR
+        logger.info(f"Rollout {req.metadata.rollout_id} completed", extra={"status": Status.rollout_finished()})
+
+    except Exception as e:
+        # Log error with structured status - THIS IS WHAT RemoteRolloutProcessor POLLS FOR
+        logger.error(
+            f"Rollout {req.metadata.rollout_id} failed: {e}", extra={"status": Status.rollout_error_from_exception(e)}
         )
 
-    def do_OPTIONS(self):
-        """Handle CORS preflight requests"""
-        self.send_response(200)
-        self.send_header("Access-Control-Allow-Origin", "*")
-        self.send_header("Access-Control-Allow-Methods", "POST, GET, OPTIONS")
-        self.send_header("Access-Control-Allow-Headers", "Content-Type")
-        self.end_headers()
-
-    def _send_json_response(self, status_code: int, data: dict):
-        """Send a JSON response"""
-        self.send_response(status_code)
-        self.send_header("Content-Type", "application/json")
-        self.send_header("Access-Control-Allow-Origin", "*")
-        self.send_header("Access-Control-Allow-Methods", "POST, GET, OPTIONS")
-        self.send_header("Access-Control-Allow-Headers", "Content-Type")
-        self.end_headers()
-        self.wfile.write(json.dumps(data).encode("utf-8"))
-
-    def _send_error(self, status_code: int, message: str):
-        """Send an error response"""
-        self._send_json_response(status_code, {"error": message})
+
+@app.route("/init", methods=["POST"])
+async def init():
+    try:
+        # Parse as InitRequest
+        req = InitRequest(**request.get_json())
+
+        # Create logger for immediate validation logging
+        logger = logging.getLogger(f"{__name__}.{req.metadata.rollout_id}")
+        logger.addFilter(RolloutIdFilter(req.metadata.rollout_id))
+
+        # Validate required fields
+        if not req.messages:
+            error_msg = "messages is required"
+            logger.error(error_msg, extra={"status": Status.rollout_error(error_msg)})
+            return jsonify({"error": error_msg}), 400
+
+        # Get API key (prefer request api_key, fallback to environment)
+        if req.api_key:
+            logger.info("Using API key from request")
+            api_key = req.api_key
+        elif os.environ.get("FIREWORKS_API_KEY"):
+            logger.info("Using API key from environment")
+            api_key = os.environ.get("FIREWORKS_API_KEY")
+        else:
+            error_msg = "API key not provided in request or environment variable"
+            logger.error(error_msg, extra={"status": Status.rollout_error(error_msg)})
+            return jsonify({"error": error_msg}), 401
+
+        # 🔥 FIRE: Return immediately with acceptance (within 30s requirement)
+        response_data = {
+            "status": "accepted",
+            "rollout_id": req.metadata.rollout_id,
+            "message": "Rollout processing started",
+        }
+
+        # Fire and forget: Execute rollout asynchronously
+        asyncio.create_task(execute_rollout_background(req, api_key))
+
+        return jsonify(response_data), 200
+
+    except Exception as e:
+        # For request parsing errors, return error immediately (don't retry)
+        return jsonify({"error": f"Request parsing error: {str(e)}"}), 400
+
+
+@app.route("/", methods=["GET"])
+def health_check():
+    """Health check endpoint"""
+    return jsonify(
+        {
+            "status": "ok",
+            "message": "SVGBench Vercel Serverless Function",
+            "endpoints": {"POST /": "Process SVGBench evaluation requests"},
+        }
+    )
+
+
+@app.route("/", methods=["OPTIONS"])
+def options_handler():
+    """Handle CORS preflight requests"""
+    response = jsonify({})
+    response.headers["Access-Control-Allow-Origin"] = "*"
+    response.headers["Access-Control-Allow-Methods"] = "POST, GET, OPTIONS"
+    response.headers["Access-Control-Allow-Headers"] = "Content-Type"
+    return response
+
+
+# Add CORS headers to all responses
+@app.after_request
+def add_cors_headers(response):
+    response.headers["Access-Control-Allow-Origin"] = "*"
+    response.headers["Access-Control-Allow-Methods"] = "POST, GET, OPTIONS"
+    response.headers["Access-Control-Allow-Headers"] = "Content-Type"
+    return response
