Handle errors from model for comfyui-json

2025-10-08 12:00:45 -07:00
11 changed files with 142 additions and 935 deletions
@@ -12,7 +12,6 @@ from distutils.util import strtobool

 from anyio import open_file
 from aiohttp import web, ClientResponse, ClientSession, ClientConnectorError, ClientTimeout, TCPConnector
-import asyncio

 import requests
 from Crypto.Signature import pkcs1_15
@@ -26,12 +25,8 @@ from lib.data_types import (
    LogAction,
    ApiPayload_T,
    JsonDataException,
-    RequestMetrics,
-    BenchmarkResult
 )

-VERSION = "0.1.0"
-
 MSG_HISTORY_LEN = 100
 log = logging.getLogger(__file__)

@@ -58,9 +53,7 @@ class Backend:
        EndpointHandler  # this endpoint handler will be used for benchmarking
    )
    log_actions: List[Tuple[LogAction, str]]
-    max_wait_time: float = 10.0
    reqnum = -1
-    version = VERSION
    msg_history = []
    sem: Semaphore = dataclasses.field(default_factory=Semaphore)
    unsecured: bool = dataclasses.field(
@@ -69,7 +62,6 @@ class Backend:

    def __post_init__(self):
        self.metrics = Metrics()
-        self.metrics._set_version(self.version)
        self._total_pubkey_fetch_errors = 0
        self._pubkey = self._fetch_pubkey()
        self.__start_healthcheck: bool = False
@@ -136,56 +128,55 @@ class Backend:
        except json.JSONDecodeError:
            return web.json_response(dict(error="invalid JSON"), status=422)
        workload = payload.count_workload()
-        request_metrics: RequestMetrics = RequestMetrics(request_idx=auth_data.request_idx, reqnum=auth_data.reqnum, workload=workload, status="Created")

        async def cancel_api_call_if_disconnected() -> web.Response:
            await request.wait_for_disconnection()
-            log.debug(f"request with reqnum: {request_metrics.reqnum} was canceled")
-            self.metrics._request_canceled(request_metrics)
-            raise asyncio.CancelledError
+            log.debug(f"request with reqnum: {auth_data.reqnum} was canceled")
+            self.metrics._request_canceled(workload=workload)
+            return web.Response(status=500)

        async def make_request() -> Union[web.Response, web.StreamResponse]:
+            log.debug(f"got request, {auth_data.reqnum}")
+            self.metrics._request_start(workload=workload, reqnum=auth_data.reqnum)
+            if self.allow_parallel_requests is False:
+                log.debug(f"Waiting to aquire Sem for reqnum:{auth_data.reqnum}")
+                await self.sem.acquire()
+                log.debug(
+                    f"Sem acquired for reqnum:{auth_data.reqnum}, starting request..."
+                )
+            else:
+                log.debug(f"Starting request for reqnum:{auth_data.reqnum}")
            try:
                response = await self.__call_api(handler=handler, payload=payload)
                status_code = response.status
                log.debug(
                    " ".join(
                        [
-                            f"request with reqnum:{request_metrics.reqnum}",
+                            f"request with reqnum:{auth_data.reqnum}",
                            f"returned status code: {status_code},",
                        ]
                    )
                )
                res = await handler.generate_client_response(request, response)
-                self.metrics._request_success(request_metrics)
+                self.metrics._request_success(workload=workload)
                return res
            except requests.exceptions.RequestException as e:
                log.debug(f"[backend] Request error: {e}")
-                self.metrics._request_errored(request_metrics)
+                self.metrics._request_errored(workload=workload)
                return web.Response(status=500)
+            finally:
+                self.metrics._request_end(
+                    workload=workload,
+                    reqnum=auth_data.reqnum,
+                )
+                self.sem.release()

        ###########

        if self.__check_signature(auth_data) is False:
-            self.metrics._request_reject(request_metrics)
            return web.Response(status=401)

-        if self.metrics.model_metrics.wait_time > self.max_wait_time:
-            self.metrics._request_reject(request_metrics)
-            return web.Response(status=429)
-
-        acquired = False
        try:
-            self.metrics._request_start(request_metrics)
-            if self.allow_parallel_requests is False:
-                log.debug(f"Waiting to aquire Sem for reqnum:{request_metrics.reqnum}")
-                await self.sem.acquire()
-                acquired = True
-                log.debug(
-                    f"Sem acquired for reqnum:{request_metrics.reqnum}, starting request..."
-                )
-            else:
-                log.debug(f"Starting request for reqnum:{request_metrics.reqnum}")
            done, pending = await wait(
                [
                    create_task(make_request()),
@@ -193,27 +184,11 @@ class Backend:
                ],
                return_when=FIRST_COMPLETED,
            )
-            for t in pending:
-                t.cancel()
-            await asyncio.gather(*pending, return_exceptions=True)
-
-            done_task = done.pop()
-            try:
-                return done_task.result()
-            except Exception as e:
-                log.debug(f"Request task raised exception: {e}")
-                return web.Response(status=500)
-        except asyncio.CancelledError:
-            # Client is gone. Do not write a response; just unwind.
-            return web.Response(status=499) 
+            [task.cancel() for task in pending]
+            return done.pop().result()
        except Exception as e:
            log.debug(f"Exception in main handler loop {e}")
            return web.Response(status=500)
-        finally:
-            # Always release the semaphore if it was acquired
-            if acquired:
-                self.sem.release()
-            self.metrics._request_end(request_metrics)

    @cached_property  
    def healthcheck_session(self):
@@ -254,7 +229,7 @@ class Backend:

    async def _start_tracking(self) -> None:
        await gather(
-            self.__read_logs(), self.metrics._send_metrics_loop(), self.__healthcheck(), self.metrics._send_delete_requests_loop()
+            self.__read_logs(), self.metrics._send_metrics_loop(), self.__healthcheck()
        )

    def backend_errored(self, msg: str) -> None:
@@ -333,26 +308,18 @@ class Backend:

            for run in range(1, self.benchmark_handler.benchmark_runs + 1):
                start = time.time()
-                benchmark_requests = []
+                tasks = []
+                total_workload = 0

-                for i in range(concurrent_requests):
+                for _ in range(concurrent_requests):
                    payload = self.benchmark_handler.make_benchmark_payload()
-                    workload = payload.count_workload()
-                    task = self.__call_api(handler=self.benchmark_handler, payload=payload)
-                    benchmark_requests.append(
-                        BenchmarkResult(request_idx=i, workload=workload, task=task)
+                    total_workload += payload.count_workload()
+                    tasks.append(
+                        self.__call_api(handler=self.benchmark_handler, payload=payload)
                    )

-                responses = await gather(*[br.task for br in benchmark_requests])
-                for br, response in zip(benchmark_requests, responses):
-                    br.response = response
-
-                total_workload = sum(br.workload for br in benchmark_requests if br.is_successful)
+                responses = await gather(*tasks)
                time_elapsed = time.time() - start
-                successful_responses = sum([1 for br in benchmark_requests if br.is_successful])
-                if successful_responses == 0:
-                    self.backend_errored("No successful responses from benchmark")
-                    log.debug(f"benchmark failed: {successful_responses}/{concurrent_requests} successful responses")

                throughput = total_workload / time_elapsed
                sum_throughput += throughput
@@ -366,7 +333,7 @@ class Backend:
                            f"Run: {run}, concurrent_requests: {concurrent_requests}",
                            f"Total workload: {total_workload}, time_elapsed: {time_elapsed}s",
                            f"Throughput: {throughput} workload/s",
-                            f"Successful responses: {successful_responses}/{concurrent_requests}",
+                            f"Successful responses: {len([r for r in responses if r.status == 200])}",
                            "#" * 60,
                        ]
                    )
@@ -420,7 +387,7 @@ class Backend:
                    if line:
                        await handle_log_line(line.rstrip())
                    else:
-                        await asyncio.sleep(LOG_POLL_INTERVAL)
+                        time.sleep(LOG_POLL_INTERVAL)

        ###########

@@ -3,7 +3,7 @@ import logging
 from dataclasses import dataclass, field
 from enum import Enum
 from abc import ABC, abstractmethod
-from typing import Dict, Any, Union, Tuple, Optional, Set, TypeVar, Generic, Type, Awaitable
+from typing import Dict, Any, Union, Tuple, Optional, Set, TypeVar, Generic, Type
 from aiohttp import web, ClientResponse
 import inspect

@@ -70,7 +70,6 @@ class AuthData:
    endpoint: str
    reqnum: int
    url: str
-    request_idx: int

    @classmethod
    def from_json_msg(cls, json_msg: Dict[str, Any]):
@@ -190,34 +189,13 @@ class SystemMetrics:
        self.additional_disk_usage = disk_usage - self.last_disk_usage
        self.last_disk_usage = disk_usage

-    def reset(self, expected: float | None) -> None:
+    def reset(self):
        # autoscaler excepts model_loading_time to be populated only once, when the instance has
        # finished benchmarking and is ready to receive requests. This applies to restarted instances
        # as well: they should send model_loading_time once when they are done loading
-        if self.model_loading_time == expected:
        self.model_loading_time = None


-@dataclass
-class RequestMetrics:
-    """Tracks metrics for an active request."""
-    request_idx: int
-    reqnum: int
-    workload: float
-    status: str
-    success: bool = False
-
-@dataclass
-class BenchmarkResult:
-    request_idx: int
-    workload: float
-    task: Awaitable[ClientResponse]
-    response: Optional[ClientResponse] = None
-
-    @property
-    def is_successful(self) -> bool:
-        return self.response is not None and self.response.status == 200
-
@dataclass
 class ModelMetrics:
    """Model specific metrics"""
@@ -227,14 +205,12 @@ class ModelMetrics:
    workload_received: float
    workload_cancelled: float
    workload_errored: float
-    workload_rejected: float
    # these are not
    workload_pending: float
    error_msg: Optional[str]
    max_throughput: float
    requests_recieved: Set[int] = field(default_factory=set)
-    requests_working: dict[int, RequestMetrics] = field(default_factory=dict)
-    requests_deleting: list[RequestMetrics] = field(default_factory=list)
+    requests_working: Set[int] = field(default_factory=set)
    last_update: float = field(default_factory=time.time)

    @classmethod
@@ -244,30 +220,19 @@ class ModelMetrics:
            workload_served=0.0,
            workload_cancelled=0.0,
            workload_errored=0.0,
-            workload_rejected=0.0,
            workload_received=0.0,
            error_msg=None,
            max_throughput=0.0,
        )

+    @property
+    def cur_perf(self) -> float:
+        return max(self.workload_served / (time.time() - self.last_update), 0.0)
+
    @property
    def workload_processing(self) -> float:
        return max(self.workload_received - self.workload_cancelled, 0.0)

-    @property
-    def wait_time(self) -> float:
-        if (len(self.requests_working) == 0):
-            return 0.0
-        return sum([request.workload for request in self.requests_working.values()]) / max(self.max_throughput, 0.00001)
-    
-    @property
-    def cur_load(self) -> float:
-        return sum([request.workload for request in self.requests_working.values()])
-
-    @property
-    def working_request_idxs(self) -> list[int]:
-        return [req.request_idx for req in self.requests_working.values()]
-
    def set_errored(self, error_msg):
        self.reset()
        self.error_msg = error_msg
@@ -277,20 +242,16 @@ class ModelMetrics:
        self.workload_received = 0
        self.workload_cancelled = 0
        self.workload_errored = 0
-        self.workload_rejected = 0
        self.last_update = time.time()


@dataclass
-class AutoScalerData:
+class AutoScalaerData:
    """Data that is reported to autoscaler"""

    id: int
-    version: str
    loadtime: float
    cur_load: float
-    rej_load: float
-    new_load: float
    error_msg: str
    max_perf: float
    cur_perf: float
@@ -299,7 +260,6 @@ class AutoScalerData:
    num_requests_working: int
    num_requests_recieved: int
    additional_disk_usage: float
-    working_request_idxs: list[int]
    url: str


@@ -5,14 +5,13 @@ import json
 from asyncio import sleep
 from dataclasses import dataclass, asdict, field
 from functools import cache
-import asyncio
-from aiohttp import ClientSession, ClientTimeout, TCPConnector, ClientResponseError

-from lib.data_types import AutoScalerData, SystemMetrics, ModelMetrics, RequestMetrics
+import requests
+
+from lib.data_types import AutoScalaerData, SystemMetrics, ModelMetrics
 from typing import Awaitable, NoReturn, List

 METRICS_UPDATE_INTERVAL = 1
-DELETE_REQUESTS_INTERVAL = 1

 log = logging.getLogger(__file__)

@@ -27,9 +26,7 @@ def get_url() -> str:

@dataclass
 class Metrics:
-    version: str = "0"
    last_metric_update: float = 0.0
-    last_request_served: float = 0.0
    update_pending: bool = False
    id: int = field(default_factory=lambda: int(os.environ["CONTAINER_ID"]))
    report_addr: List[str] = field(
@@ -38,84 +35,42 @@ class Metrics:
    url: str = field(default_factory=get_url)
    system_metrics: SystemMetrics = field(default_factory=SystemMetrics.empty)
    model_metrics: ModelMetrics = field(default_factory=ModelMetrics.empty)
-    _session: ClientSession | None = field(default=None, init=False, repr=False)

-    async def http(self) -> ClientSession:
-        if self._session is None:
-            self._session = ClientSession(
-                timeout=ClientTimeout(total=10),
-                connector=TCPConnector(limit=8, limit_per_host=4, force_close=True, enable_cleanup_closed=True)
-            )
-        return self._session
-    
-    async def aclose(self) -> None:
-        if self._session is not None:
-            await self._session.close()
-            self._session = None
-
-    def _request_start(self, request: RequestMetrics) -> None:
+    def _request_start(self, workload: float, reqnum: int) -> None:
        """
        this function is called prior to forwarding a request to a model API.
        """
        log.debug("request start")
-        request.status = "Started"
-        self.model_metrics.workload_pending += request.workload
-        self.model_metrics.workload_received += request.workload
-        self.model_metrics.requests_recieved.add(request.reqnum)
-        self.model_metrics.requests_working[request.reqnum] = request
-        self.update_pending = True
+        self.model_metrics.workload_pending += workload
+        self.model_metrics.workload_received += workload
+        self.model_metrics.requests_recieved.add(reqnum)
+        self.model_metrics.requests_working.add(reqnum)

-    def _request_end(self, request: RequestMetrics) -> None:
+    def _request_end(self, workload: float, reqnum: int) -> None:
        """
        this function is called after handling of a request ends, regardless of the outcome
        """
-        self.model_metrics.workload_pending -= request.workload
-        self.model_metrics.requests_working.pop(request.reqnum, None)
-        self.model_metrics.requests_deleting.append(request)
-        self.last_request_served = time.time()
+        self.model_metrics.workload_pending -= workload
+        self.model_metrics.requests_working.discard(reqnum)

-    def _request_success(self, request: RequestMetrics) -> None:
+    def _request_success(self, workload: float) -> None:
        """
        this function is called after a response from model API is received and forwarded.
        """
-        self.model_metrics.workload_served += request.workload
-        request.status = "Success"
-        request.success = True
+        self.model_metrics.workload_served += workload
        self.update_pending = True

-    def _request_errored(self, request: RequestMetrics) -> None:
+    def _request_errored(self, workload: float) -> None:
        """
        this function is called if model API returns an error
        """
-        self.model_metrics.workload_errored += request.workload
-        request.status = "Error"
-        request.success = False
-        self.update_pending = True
+        self.model_metrics.workload_errored += workload

-    def _request_canceled(self, request: RequestMetrics) -> None:
+    def _request_canceled(self, workload: float) -> None:
        """
        this function is called if client drops connection before model API has responded
        """
-        self.model_metrics.workload_cancelled += request.workload
-        request.success = True
-        request.status = "Cancelled"
-    
-    def _request_reject(self, request: RequestMetrics):
-        """
-        this function is called if the current wait time for the model is above max_wait_time
-        """
-        self.model_metrics.requests_recieved.add(request.reqnum)
-        self.model_metrics.requests_deleting.append(request)
-        self.model_metrics.workload_rejected += request.workload
-        request.success = False
-        request.status = "Rejected"
-        self.update_pending = True
-
-    async def _send_delete_requests_loop(self) -> Awaitable[NoReturn]:
-        while True:
-            await sleep(DELETE_REQUESTS_INTERVAL)
-            if len(self.model_metrics.requests_deleting) > 0:
-                await self.__send_delete_requests_and_reset()
+        self.model_metrics.workload_cancelled += workload

    async def _send_metrics_loop(self) -> Awaitable[NoReturn]:
        while True:
@@ -123,10 +78,10 @@ class Metrics:
            elapsed = time.time() - self.last_metric_update
            if self.system_metrics.model_is_loaded is False and elapsed >= 10:
                log.debug(f"sending loading model metrics after {int(elapsed)}s wait")
-                await self.__send_metrics_and_reset()
+                self.__send_metrics_and_reset(elapsed)
            elif self.update_pending or elapsed > 10:
                log.debug(f"sending loaded model metrics after {int(elapsed)}s wait")
-                await self.__send_metrics_and_reset()
+                self.__send_metrics_and_reset(elapsed)

    def _model_loaded(self, max_throughput: float) -> None:
        self.system_metrics.model_loading_time = (
@@ -139,90 +94,27 @@ class Metrics:
        self.model_metrics.set_errored(error_msg)
        self.system_metrics.model_is_loaded = True

-    def _set_version(self, version: str) -> None:
-        self.version = version
-
    #######################################Private#######################################

-    async def __send_delete_requests_and_reset(self):
-        async def post(report_addr: str, idxs: list[int], success_flag: bool) -> bool:
-            data = {
-                "worker_id": self.id,
-                "request_idxs": idxs,
-                "success": success_flag,
-            }
-            log.debug(
-                f"Deleting requests that {'succeeded' if success_flag else 'failed'}: {data['request_idxs']}"
-            )
-            full_path = report_addr.rstrip("/") + "/delete_requests/"
-            for attempt in range(1, 4):
-                try:
-                    session = await self.http()
-                    async with session.post(full_path, json=data) as res:
-                        log.debug(f"delete_requests response: {res.status}")
-                        res.raise_for_status()
-                    return True
-                except asyncio.TimeoutError:
-                    log.debug("delete_requests timed out")
-                except (ClientResponseError, Exception) as e:
-                    log.debug(f"delete_requests failed with error: {e}")
-                await asyncio.sleep(2)
-                log.debug(f"retrying delete_request, attempt: {attempt}")
-            return False
+    def __send_metrics_and_reset(self, elapsed):

-        # Take a snapshot of what we plan to send this tick.
-        # New arrivals after this snapshot will remain in the queue for the next tick.
-        snapshot = list(self.model_metrics.requests_deleting)
-        success_idxs = [r.request_idx for r in snapshot if r.success is True]
-        failed_idxs  = [r.request_idx for r in snapshot if r.success is False]
-
-        if not success_idxs and not failed_idxs:
-            return  # nothing to do
-
-        for report_addr in self.report_addr:
-            sent_success = True
-            sent_failed  = True
-
-            if success_idxs:
-                sent_success = await post(report_addr, success_idxs, True)
-            if failed_idxs:
-                sent_failed = await post(report_addr, failed_idxs, False)
-
-            if sent_success and sent_failed:
-                # Remove only the items we actually sent from the live queue.
-                sent_set = set(success_idxs) | set(failed_idxs)
-                self.model_metrics.requests_deleting[:] = [
-                    r for r in self.model_metrics.requests_deleting
-                    if r.request_idx not in sent_set
-                ]
-                break
-
-
-    async def __send_metrics_and_reset(self):
-
-        loadtime_snapshot = self.system_metrics.model_loading_time
-
-        def compute_autoscaler_data() -> AutoScalerData:
-            return AutoScalerData(
+        def compute_autoscaler_data() -> AutoScalaerData:
+            return AutoScalaerData(
                id=self.id,
-                version=self.version,
-                loadtime=(loadtime_snapshot or 0.0), 
-                new_load=self.model_metrics.workload_processing,
-                cur_load=self.model_metrics.cur_load,
-                rej_load=self.model_metrics.workload_rejected,
+                loadtime=(self.system_metrics.model_loading_time or 0.0),
+                cur_load=(self.model_metrics.workload_processing / elapsed),
                max_perf=self.model_metrics.max_throughput,
-                cur_perf=self.model_metrics.workload_served,
+                cur_perf=self.model_metrics.cur_perf,
                error_msg=self.model_metrics.error_msg or "",
                num_requests_working=len(self.model_metrics.requests_working),
                num_requests_recieved=len(self.model_metrics.requests_recieved),
                additional_disk_usage=self.system_metrics.additional_disk_usage,
-                working_request_idxs=self.model_metrics.working_request_idxs,
                cur_capacity=0,
                max_capacity=0,
                url=self.url,
            )

-        async def send_data(report_addr: str) -> bool:
+        def send_data(report_addr: str) -> bool:
            data = compute_autoscaler_data()
            full_path = report_addr.rstrip("/") + "/worker_status/"
            log.debug(
@@ -237,15 +129,14 @@ class Metrics:
            )
            for attempt in range(1, 4):
                try:
-                    session = await self.http()
-                    async with session.post(full_path, json=asdict(data)) as res:
+                    res = requests.post(full_path, json=asdict(data), timeout=1)
                    res.raise_for_status()
                    return True
-                except asyncio.TimeoutError:
+                except requests.Timeout:
                    log.debug(f"autoscaler status update timed out")
-                except (ClientResponseError, Exception)  as e:
+                except Exception as e:
                    log.debug(f"autoscaler status update failed with error: {e}")
-                await asyncio.sleep(2)
+                time.sleep(2)
                log.debug(f"retrying autoscaler status update, attempt: {attempt}")
            log.debug(f"failed to send update through {report_addr}")
            return False
@@ -254,15 +145,11 @@ class Metrics:

        self.system_metrics.update_disk_usage()

-        sent = False
        for report_addr in self.report_addr:
-            if await send_data(report_addr):
-                sent = True
+            success = send_data(report_addr)
+            if success is True:
                break
-
-        if sent:
-            # clear the one-shot loadtime only if we actually sent *this* value
-            self.system_metrics.reset(expected=loadtime_snapshot)
        self.update_pending = False
        self.model_metrics.reset()
+        self.system_metrics.reset()
        self.last_metric_update = time.time()
@@ -292,12 +292,12 @@ def test_load_cmd(
    args = arg_parser.parse_args()
    if hasattr(args, "comfy_model"):
        os.environ["COMFY_MODEL"] = args.comfy_model
-    server_url = {
-        "prod": "https://run.vast.ai",
-        "alpha": "https://run-alpha.vast.ai",
-        "candidate": "https://run-candidate.vast.ai",
-        "local": "http://localhost:8080",
-    }.get(args.instance, "http://localhost:8080")
+    server_url = dict(
+        prod="https://run.vast.ai",
+        alpha="https://run-alpha.vast.ai",
+        candidate="https://run-candidate.vast.ai",
+        local="http://localhost:8080",
+    )[args.instance]
    run_test(
        num_requests=args.num_requests,
        requests_per_second=args.requests_per_second,
@@ -1,6 +1,5 @@
 import logging
-import time
-from typing import Any, Dict, Optional, Tuple
+from typing import Any, Dict, Optional

 import requests

@@ -17,38 +16,6 @@ class Endpoint:
    Utility class for handling endpoint operations.
    """

-    @staticmethod
-    def get_endpoint_info(
-        endpoint_name: str, account_api_key: str, instance: str
-    ) -> Optional[Dict[str, Any]]:
-        headers = {"Authorization": f"Bearer {account_api_key}"}
-        url = f"{Endpoint.get_server_url(instance)}?autoscaler_instance={instance}"
-        # Retry a few times to smooth over transient propagation/network delays
-        for attempt in range(4):
-            try:
-                response = requests.get(url, headers=headers, timeout=8)
-                if response.status_code != 200:
-                    # brief backoff and retry
-                    time.sleep(0.3 * (attempt + 1))
-                    continue
-                try:
-                    data = response.json()
-                except Exception:
-                    # JSON parse failed; backoff and retry
-                    time.sleep(0.3 * (attempt + 1))
-                    continue
-                result = data.get("results", []) if isinstance(data, dict) else []
-                endpoint = next(
-                    (item for item in result if item.get("endpoint_name") == endpoint_name),
-                    None,
-                )
-                if endpoint and endpoint.get("id") and endpoint.get("api_key"):
-                    return {"id": endpoint.get("id"), "api_key": endpoint.get("api_key")}
-            except Exception:
-                # network or other transient error; retry
-                time.sleep(0.3 * (attempt + 1))
-        return None
-
    @staticmethod
    def get_autoscaler_server_url(instance: str) -> str:
        endpoints = {
@@ -56,10 +23,7 @@ class Endpoint:
            "candidate": "run-candidate",
            "prod": "run",
        }
-        host = endpoints.get(instance)
-        if host:
-            return f"https://{host}.vast.ai/"
-        return "http://localhost:8080"
+        return f"https://{endpoints[instance]}.vast.ai/"

    @staticmethod
    def get_server_url(instance: str) -> str:
@@ -68,8 +32,7 @@ class Endpoint:
            "candidate": "candidate",
            "prod": "console",
        }
-        host = endpoints.get(instance, "alpha")
-        return f"https://{host}.vast.ai/api/v0/endptjobs/"
+        return f"https://{endpoints[instance]}.vast.ai/api/v0/endptjobs/"

    @staticmethod
    def get_endpoint_api_key(
@@ -92,7 +55,6 @@ class Endpoint:
            response = requests.get(
                f"{Endpoint.get_server_url(instance)}?autoscaler_instance={instance}",
                headers=headers,
-                timeout=8,
            )

            if response.status_code != 200:
@@ -102,14 +64,14 @@ class Endpoint:

            try:
                data = response.json()
-            except Exception as e:
+            except requests.exceptions.JSONDecodeError as e:
                log.debug(f"Failed to parse JSON response: {e}")
                return None

            result = data.get("results", [])

            endpoint: Optional[Dict[str, Any]] = next(
-                (item for item in result if item.get("endpoint_name") == endpoint_name),
+                (item for item in result if item["endpoint_name"] == endpoint_name),
                None,
            )
            if not endpoint:
@@ -12,21 +12,9 @@ A docker image is provided but you may use any if the above requirements are met

 ## Benchmarking

-### Custom Benchmark Workflows
+A simple image generation benchmark runs when each worker initializes to validate GPU performance and identify underperforming machines.

-You can provide a custom ComfyUI workflow for benchmarking by creating `workers/comfyui-json/misc/benchmark.json`. This allows you to test performance using your preferred models and workflow complexity.
-
-**Ways to provide the benchmark file:**
- Fork this repository and add your `benchmark.json` file
- Write the file during worker provisioning (onstart script or setup phase)
-
-An example file is provided in the repository. To ensure varied generations, use the placeholder `__RANDOM_INT__` in place of static seed values - it will be replaced with a random integer for each benchmark run.
-
-### Default Benchmark (Fallback)
-
-If `benchmark.json` is not available, a simple image generation benchmark runs when each worker initializes. This validates GPU performance and helps identify underperforming machines.
-
-The default benchmark uses Stable Diffusion v1.5 with ComfyUI's standard text-to-image workflow. Configure it using these environment variables:
+The benchmark uses Stable Diffusion v1.5 with ComfyUI's default text-to-image workflow. Configure the benchmark complexity and duration using these variables:

 | Environment Variable | Default Value | Description |
 | -------------------- | ------------- | ----------- |
@@ -36,7 +24,7 @@ The default benchmark uses Stable Diffusion v1.5 with ComfyUI's standard text-to

 Each benchmark run uses a random prompt from `misc/test_prompts.txt` and a random seed to ensure consistent GPU load patterns.

-#### Calibrating Fallback Benchmark Duration
+### Calibrating Benchmark Duration

 To screen for underperforming hardware, set `BENCHMARK_TEST_STEPS` to match your expected production workflow duration. This allows you to identify machines that won't meet performance requirements.

@@ -5,13 +5,12 @@ import dataclasses
 from typing import Dict, Any
 from functools import cache
 from math import ceil
-from pathlib import Path
-import json
-import logging

 from lib.data_types import ApiPayload, JsonDataException

-log = logging.getLogger(__file__)
+
+with open("workers/comfyui/misc/test_prompts.txt", "r") as f:
+    test_prompts = f.readlines()

 def count_workload() -> float:
    # Always 100.0 where there is a single instance of ComfyUI handling requests
@@ -25,32 +24,9 @@ class ComfyWorkflowData(ApiPayload):
    @classmethod
    def for_test(cls):
        """
-        If the user has provided a benchmark workflow we can use it here to properly gauge performance.
-        Otherwise, use the variables available to simulate workflows of the required running time
+        Use the variables available to simulate workflows of the required running time
        Example: SD1.5, simple image gen 10000 steps, 512px x 512px will run for approximately 9 minutes @ ~18 it/s (RTX 4090)
        """
-        # Try to load benchmark.json
-        benchmark_file = Path("workers/comfyui-json/misc/benchmark.json")
-        
-        if benchmark_file.exists():
-            try:
-                with open(benchmark_file, "r") as f:
-                    benchmark_workflow = json.load(f)
-                return cls(
-                    input={
-                        "request_id": f"test-{random.randint(1000, 99999)}",
-                        "workflow_json": benchmark_workflow
-                    }
-                )
-            except (json.JSONDecodeError, IOError):
-                # JSON is malformed or file can't be read, fall through to default
-                log.error(f"Failed to benchmark using {benchmark_file}")
-        
-        # Fallback: read prompts and construct payload
-        log.info("Using fallback method for benchmarking")
-        with open("workers/comfyui-json/misc/test_prompts.txt", "r") as f:
-            test_prompts = f.readlines()
-        
        test_prompt = random.choice(test_prompts).rstrip()
        return cls(
            input={
@@ -1,107 +0,0 @@
-{
-    "3": {
-        "inputs": {
-            "seed": "__RANDOM_INT__",
-            "steps": 20,
-            "cfg": 8,
-            "sampler_name": "euler",
-            "scheduler": "normal",
-            "denoise": 1,
-            "model": [
-            "4",
-            0
-            ],
-            "positive": [
-            "6",
-            0
-            ],
-            "negative": [
-            "7",
-            0
-            ],
-            "latent_image": [
-            "5",
-            0
-            ]
-        },
-        "class_type": "KSampler",
-        "_meta": {
-            "title": "KSampler"
-        }
-    },
-    "4": {
-        "inputs": {
-            "ckpt_name": "v1-5-pruned-emaonly-fp16.safetensors"
-        },
-        "class_type": "CheckpointLoaderSimple",
-        "_meta": {
-            "title": "Load Checkpoint"
-        }
-    },
-    "5": {
-        "inputs": {
-            "width": 512,
-            "height": 512,
-            "batch_size": 1
-        },
-        "class_type": "EmptyLatentImage",
-        "_meta": {
-            "title": "Empty Latent Image"
-        }
-    },
-    "6": {
-        "inputs": {
-            "text": "beautiful scenery nature glass bottle landscape, , purple galaxy bottle,",
-            "clip": [
-            "4",
-            1
-            ]
-        },
-        "class_type": "CLIPTextEncode",
-        "_meta": {
-            "title": "CLIP Text Encode (Prompt)"
-        }
-    },
-    "7": {
-        "inputs": {
-            "text": "text, watermark",
-            "clip": [
-            "4",
-            1
-            ]
-        },
-        "class_type": "CLIPTextEncode",
-        "_meta": {
-            "title": "CLIP Text Encode (Prompt)"
-        }
-    },
-    "8": {
-        "inputs": {
-            "samples": [
-            "3",
-            0
-            ],
-            "vae": [
-            "4",
-            2
-            ]
-        },
-        "class_type": "VAEDecode",
-        "_meta": {
-            "title": "VAE Decode"
-        }
-    },
-    "9": {
-        "inputs": {
-            "filename_prefix": "ComfyUI",
-            "images": [
-            "8",
-            0
-            ]
-        },
-        "class_type": "SaveImage",
-        "_meta": {
-            "title": "Save Image"
-        }
-    }
-}
@@ -19,7 +19,6 @@ MODEL_SERVER_START_LOG_MSG = "To see the GUI go to: "
 MODEL_SERVER_ERROR_LOG_MSGS = [
    "MetadataIncompleteBuffer",  # This error is emitted when the downloaded model is corrupted
    "Value not in list: ",  # This error is emitted when the model file is not there at all
-    "[ERROR] Provisioning Script failed", # Error inserted by provisioning script if models/nodes fail to download
 ]


@@ -34,6 +33,9 @@ log = logging.getLogger(__file__)
 async def generate_client_response(
        client_request: web.Request, model_response: ClientResponse
    ) -> Union[web.Response, web.StreamResponse]:
+        match model_response.status:
+            case 200:
+                log.debug("SUCCESS")
                # Check if the response is actually streaming based on response headers/content-type
                is_streaming_response = (
                    model_response.content_type == "text/event-stream"
@@ -60,6 +62,9 @@ async def generate_client_response(
                        status=model_response.status,
                        content_type=model_response.content_type
                    )
+            case code:
+                log.debug(f"Model responded with error {code}")
+                return web.Response(status=code)
                

@dataclasses.dataclass
@@ -119,25 +119,14 @@ class GenericHandler(EndpointHandler[GenericData], ABC):
 class CompletionsData(GenericData):
    @classmethod
    def for_test(cls) -> "CompletionsData":
-        system_prompt = """You are a helpful AI assistant. You have access to the following knowledge base:
-    
-        Zebras (US: /ˈziːbrəz/, UK: /ˈzɛbrəz, ˈziː-/)[2] (subgenus Hippotigris) are African equines 
-        with distinctive black-and-white striped coats. There are three living species: Grévy's zebra 
-        (Equus grevyi), the plains zebra (E. quagga), and the mountain zebra (E. zebra). Zebras share the 
-        genus Equus with horses and asses, the three groups being the only living members of the family 
-        Equidae. Zebra stripes come in different patterns, unique to each individual. Zebras inhabit eastern 
-        and southern Africa and can be found in a variety of habitats such as savannahs, grasslands, 
-        woodlands, shrublands, and mountainous areas.
-        
-        Please answer the following question based on the above context."""
-        unique_question = " ".join(random.choices(WORD_LIST, k=int(100)))
+        prompt = " ".join(random.choices(WORD_LIST, k=int(250)))
        model = os.environ.get("MODEL_NAME")
        if not model:
            raise ValueError("MODEL_NAME environment variable not set")

        test_input = {
            "model": model,
-            "prompt": f"{system_prompt}\n\n{unique_question}",
+            "prompt": prompt,
            "temperature": 0.7,
            "max_tokens": 500,
        }
@@ -164,18 +153,7 @@ class ChatCompletionsData(GenericData):

    @classmethod
    def for_test(cls) -> "ChatCompletionsData":
-        system_prompt = """You are a helpful AI assistant. You have access to the following knowledge base:
-    
-        Zebras (US: /ˈziːbrəz/, UK: /ˈzɛbrəz, ˈziː-/)[2] (subgenus Hippotigris) are African equines 
-        with distinctive black-and-white striped coats. There are three living species: Grévy's zebra 
-        (Equus grevyi), the plains zebra (E. quagga), and the mountain zebra (E. zebra). Zebras share the 
-        genus Equus with horses and asses, the three groups being the only living members of the family 
-        Equidae. Zebra stripes come in different patterns, unique to each individual. Zebras inhabit eastern 
-        and southern Africa and can be found in a variety of habitats such as savannahs, grasslands, 
-        woodlands, shrublands, and mountainous areas.
-        
-        Please answer the following question based on the above context."""
-        unique_question = " ".join(random.choices(WORD_LIST, k=int(100)))
+        prompt = " ".join(random.choices(WORD_LIST, k=int(250)))
        model = os.environ.get("MODEL_NAME")
        if not model:
            raise ValueError("MODEL_NAME environment variable not set")
@@ -183,10 +161,7 @@ class ChatCompletionsData(GenericData):
        # Chat completions use messages format instead of prompt
        test_input = {
            "model": model,
-            "messages": [
-                {"role": "system", "content": system_prompt},  # Shared prefix
-                {"role": "user", "content": unique_question}   # Unique per request
-            ],
+            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.7,
            "max_tokens": 500,
        }
@@ -1,395 +1,8 @@
-from lib.test_utils import test_args
-from utils.endpoint_util import Endpoint
-from utils.ssl import get_cert_file_path
-from lib.data_types import AuthData
+from lib.test_utils import test_load_cmd, test_args
 from .data_types.server import CompletionsData
-
 import os
-import time
-import threading
-import requests
-from dataclasses import dataclass
-from collections import Counter
-from urllib.parse import urljoin, urlparse
-import re

-# Headless plotting
-import matplotlib
-matplotlib.use("Agg")
-import logging
-logging.getLogger("matplotlib.font_manager").setLevel(logging.WARNING)
-import matplotlib.pyplot as plt
-import numpy as np
-from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
-from requests.adapters import HTTPAdapter
-
-def get_incremented_path(path: str) -> str:
-    base, ext = os.path.splitext(path)
-    if not os.path.exists(path):
-        return path
-    i = 1
-    while os.path.exists(f"{base}-{i}{ext}"):
-        i += 1
-    return f"{base}-{i}{ext}"
-
-WORKER_ENDPOINT = "/v1/completions" # This will return the full text output at once. Latency metrics reflect that (ie not measuring TTFT)
-
-@dataclass
-class ReqResult:
-    worker_url: str
-    route_ms: float
-    worker_ms: float
-    total_ms: float
-    ok: bool
-    error: str = ""
-    status_code: int = 0
-    t_start: float = 0.0
-    t_end: float = 0.0
-    workload: float = 0.0
-
-def do_one(endpoint_name: str,
-           endpoint_id: int,
-           endpoint_api_key: str,
-           server_url: str,
-           worker_endpoint: str,
-           payload,
-           results_list,
-           t0,
-           status_samples,
-           route_session,
-           worker_session):
-    try:
-        workload = payload.count_workload()
-        route_payload = {"endpoint": endpoint_name, "api_key": endpoint_api_key, "cost": workload}
-        headers = {"Authorization": f"Bearer {endpoint_api_key}"}
-        start = time.time()
-        r0 = route_session.post(urljoin(server_url, "/route/"), json=route_payload, headers=headers, timeout=4)
-        t_after_route = time.time()
-        if r0.status_code != 200:
-            results_list.append(ReqResult(worker_url="", 
-                                            route_ms=(t_after_route - start) * 1000.0, 
-                                            worker_ms=0.0, 
-                                            total_ms=(t_after_route - start) * 1000.0, 
-                                            ok=False, 
-                                            error=f"route error {r0.reason} {r0.text}",
-                                            status_code=r0.status_code,
-                                            t_start=start - t0, 
-                                            t_end=t_after_route - t0, 
-                                            workload=workload))
-            return
-        msg = r0.json()
-
-        # 1) Check if we got a worker back from route
-        worker_url = msg.get("url", "")
-        if not worker_url:
-            status = msg.get("status", "")
-            m = re.search(r"total workers:\s*(\d+).*loading workers:\s*(\d+).*standby workers:\s*(\d+).*error workers:\s*(\d+)", status, re.I | re.S)
-            if m:
-                tot, loading, standby, err = map(int, m.groups())
-                idle = max(tot - loading - standby - err, 0)
-                status_samples.append((time.time() - t0, idle))
-
-        # 2) If we got a worker, send the request
-        if worker_url:
-            req = dict(payload=payload.__dict__, auth_data=AuthData.from_json_msg(msg).__dict__)
-            t_before_worker = time.time()
-            r1 = worker_session.post(
-                urljoin(worker_url, worker_endpoint),
-                json=req,
-                verify=get_cert_file_path(),
-                timeout=(4, 120),
-            )
-            t_after_worker = time.time()
-            if r1.status_code != 200:
-                results_list.append(ReqResult(worker_url=worker_url, 
-                                                route_ms=(t_after_route - start) * 1000.0, 
-                                                worker_ms=(t_after_worker - t_before_worker) * 1000.0,
-                                                total_ms=(t_after_worker - start) * 1000.0, 
-                                                ok=False,
-                                                error=f"worker inference error {r1.reason} {r1.text}",
-                                                status_code=r1.status_code,
-                                                t_start=start - t0, 
-                                                t_end=t_after_worker - t0, 
-                                                workload=workload))
-                return
-            # Success case
-            results_list.append(ReqResult(worker_url=worker_url, 
-                                            route_ms=(t_after_route - start) * 1000.0, 
-                                            worker_ms=(t_after_worker - t_before_worker) * 1000.0, 
-                                            total_ms=(t_after_worker - start) * 1000.0,
-                                            ok=True,
-                                            error="",
-                                            status_code=200,
-                                            t_start=start - t0, 
-                                            t_end=t_after_worker - t0, 
-                                            workload=workload))
-
-        # 3) If so, sample via /get_endpoint_workers/ for eligible (idle) worker tracking
-        if worker_url:
-            try:
-                r_status = route_session.post(
-                    urljoin(server_url, "/get_endpoint_workers/"),
-                    json={"id": endpoint_id},
-                    headers={"Authorization": f"Bearer {endpoint_api_key}"},
-                    timeout=3,
-                )
-                if r_status.status_code == 200:
-                    workers = r_status.json()
-                    idle = 0
-                    for w in workers:
-                        st = str(w.get("status", "")).lower()
-                        if (st in ("idle")):
-                            idle += 1
-                    status_samples.append((time.time() - t0, idle))
-            except Exception:
-                pass
-    except Exception as e:
-        t = time.time()
-        results_list.append(ReqResult(worker_url="", 
-                                        route_ms=0.0, 
-                                        worker_ms=0.0, 
-                                        total_ms=0.0, 
-                                        ok=False, 
-                                        error=f"unknown error {e}",
-                                        status_code=0,
-                                        t_start=t - t0, 
-                                        t_end=t - t0, 
-                                        workload=0.0))
-
-def run_load_with_metrics(num_requests: int,
-                          requests_per_second: float,
-                          endpoint_group_name: str,
-                          account_api_key: str,
-                          server_url: str,
-                          worker_endpoint: str,
-                          instance: str,
-                          out_path: str):
-
-    ep_info = Endpoint.get_endpoint_info(endpoint_name=endpoint_group_name,
-                                         account_api_key=account_api_key,
-                                         instance=instance)
-    if not ep_info or not ep_info.get("api_key") or not ep_info.get("id"):
-        print(f"Endpoint {endpoint_group_name} not found for API key")
-        return
-    endpoint_id = int(ep_info["id"])
-    endpoint_api_key = ep_info["api_key"]
-
-    t0 = time.time()
-    results = []
-    status_samples = []
-    max_concurrency = int(os.environ.get("MAX_CONCURRENCY", "8192"))
-    submit_queue_factor = 2  # cap queued tasks to reduce memory
-
-    # Shared HTTP sessions with connection pooling (persistent connections)
-    def make_session(pool_connections: int, pool_maxsize: int) -> requests.Session:
-        sess = requests.Session()
-        adapter = HTTPAdapter(pool_connections=pool_connections, pool_maxsize=pool_maxsize, max_retries=0)
-        sess.mount("https://", adapter)
-        sess.mount("http://", adapter)
-        return sess
-
-    # Router: mostly single host, small connection pool is sufficient
-    route_session = make_session(pool_connections=1, pool_maxsize=max_concurrency)
-    # Workers: many hosts; allow many pools and per-host concurrency up to max_concurrency
-    worker_session = make_session(pool_connections=64, pool_maxsize=max_concurrency // 8)
-
-    # Fire requests using a thread pool, scheduling at requested RPS
-    inflight = set()
-    with ThreadPoolExecutor(max_workers=max_concurrency) as executor:
-        for i in range(num_requests):
-            # Pace submissions to RPS
-            target_time = t0 + i / max(requests_per_second, 1e-9)
-            sleep_s = target_time - time.time()
-            if sleep_s > 0:
-                time.sleep(min(sleep_s, 0.5))  # sleep in chunks to stay responsive
-
-            payload = CompletionsData.for_test()
-            fut = executor.submit(
-                do_one,
-                endpoint_group_name,
-                endpoint_id,
-                endpoint_api_key,
-                server_url,
-                worker_endpoint,
-                payload,
-                results,
-                t0,
-                status_samples,
-                route_session,
-                worker_session,
-            )
-            inflight.add(fut)
-            # Prevent unbounded queue growth
-            if len(inflight) >= max_concurrency * submit_queue_factor:
-                done, not_done = wait(inflight, return_when=FIRST_COMPLETED)
-                inflight = not_done
-        # Wait for all outstanding tasks
-        if inflight:
-            wait(inflight)
-    # Close sessions
-    try:
-        route_session.close()
-    finally:
-        worker_session.close()
-
-    # Aggregate results
-    oks = [r for r in results if r.ok]
-    errs = [r for r in results if not r.ok]
-    total_reqs = len(results)
-    succ = len(oks)
-
-    total_ms = np.array([r.total_ms for r in oks]) if succ else np.array([])
-    worker_ms = np.array([r.worker_ms for r in oks]) if succ else np.array([])
-    route_ms = np.array([r.route_ms for r in oks]) if succ else np.array([])
-
-    avg_total = float(np.mean(total_ms)) if succ else 0.0
-    avg_worker = float(np.mean(worker_ms)) if succ else 0.0
-    avg_route = float(np.mean(route_ms)) if succ else 0.0
-    p50_total, p95_total = (float(np.percentile(total_ms, 50)), float(np.percentile(total_ms, 95))) if succ else (0.0, 0.0)
-
-    # Distribution over workers (by host:port)
-    hosts = [urlparse(r.worker_url).netloc for r in oks if r.worker_url]
-    dist = Counter(hosts)
-
-    # Idle over time (mode per second)
-    idle_ts, idle_vals = [], []
-    if status_samples:
-        buckets = {}
-        for ts, idle in status_samples:
-            k = int(ts)
-            buckets.setdefault(k, []).append(idle)
-        keys = sorted(buckets.keys())
-        idle_ts = keys
-        # Use the most frequent sampled value per second (mode) to keep integer counts
-        idle_vals = []
-        for k in keys:
-            vals_k = [int(v) for v in buckets[k]]
-            if vals_k:
-                cnt = Counter(vals_k)
-                idle_vals.append(cnt.most_common(1)[0][0])
-            else:
-                idle_vals.append(0)
-
-    print(f"\nResults: total={total_reqs} success={succ} errors={len(errs)}")
-    print(f"Avg latency (ms): {avg_total:.1f}  p50: {p50_total:.1f}  p95: {p95_total:.1f}")
-    print(f"Avg route latency (ms): {avg_route:.1f}  Avg worker latency (ms): {avg_worker:.1f}")
-    if errs:
-        print("Sample errors:")
-        for e in errs[:5]:
-            print(f"  {e.status_code} {e.error}")
-
-    # Plot: 2x3 grid
-    fig, axes = plt.subplots(2, 3, figsize=(15, 8))
-    fig.suptitle(f"Load test: {endpoint_group_name}  n={total_reqs}, rps={requests_per_second}, success={succ}")
-
-    # Dist per worker
-    ax0 = axes[0, 0]
-    if dist:
-        items = sorted(dist.items(), key=lambda kv: kv[1], reverse=True)
-        labels, counts = zip(*items)
-        ax0.bar(range(len(labels)), counts)
-        ax0.set_xticks(range(len(labels)))
-        ax0.set_xticklabels(labels, rotation=45, ha="right", fontsize=8)
-    ax0.set_title("Request distribution over workers")
-    ax0.set_ylabel("count")
-
-    # Latency histogram (total)
-    ax1 = axes[0, 1]
-    if succ:
-        ax1.hist(total_ms, bins=30)
-    ax1.set_title("Total latency (ms)")
-    ax1.set_xlabel("ms")
-    ax1.set_ylabel("freq")
-
-    # Eligible workers over time
-    ax_idle = axes[0, 2]
-    if idle_ts:
-        ax_idle.plot(idle_ts, idle_vals, "-o", ms=3)
-    ax_idle.set_title("Eligible workers over time")
-    ax_idle.set_xlabel("time (s)")
-    ax_idle.set_ylabel("eligible count")
-
-    # Throughput over time (completions/sec)
-    ax_idle = axes[1, 0]
-    ax_idle.clear()
-    if succ:
-        per_sec = {}
-        for r in oks:
-            s = int(r.t_end)
-            per_sec[s] = per_sec.get(s, 0) + 1
-        ts = sorted(per_sec.keys())
-        vals = [per_sec[t] for t in ts]
-        ax_idle.plot(ts, vals, "-o", ms=3)
-    ax_idle.set_title("Completions per second")
-    ax_idle.set_xlabel("time (s)")
-    ax_idle.set_ylabel("completions / sec")
-
-    # Summary text
-    ax3 = axes[1, 1]
-    ax3.axis("off")
-    text = (
-        f"Total requests: {total_reqs}\n"
-        f"Success: {succ}  Errors: {len(errs)}\n"
-        f"Avg total latency: {avg_total:.1f} ms\n"
-        f"p50: {p50_total:.1f} ms  p95: {p95_total:.1f} ms\n"
-        f"Avg route latency: {avg_route:.1f} ms\n"  
-        f"Avg worker latency: {avg_worker:.1f} ms\n"
-        f"300 errors: {len([r for r in errs if r.status_code >= 300 and r.status_code < 400])}\n"
-        f"429 errors: {len([r for r in errs if r.status_code == 429])}\n"
-        f"500 errors: {len([r for r in errs if r.status_code >= 500])}\n"
-        f"Other errors: {len([r for r in errs if r.status_code not in [300, 429, 500]])}\n"
-    )
-    ax3.set_title("Summary")
-    ax3.text(0.02, 0.98, text, va="top", ha="left", fontsize=11, transform=ax3.transAxes)
-
-    # Error count over time
-    ax_errors = axes[1, 2]
-    all_end_times = [int(r.t_end) for r in results if r.t_end > 0]
-    if all_end_times:
-        min_second = min(all_end_times)
-        max_second = max(all_end_times)
-        # Count errors per second
-        errors_per_second = {}
-        for result in errs:
-            second = int(result.t_end)
-            errors_per_second[second] = errors_per_second.get(second, 0) + 1
-        # Create complete timeline including zeros
-        time_seconds = list(range(min_second, max_second + 1))
-        error_counts = [errors_per_second.get(sec, 0) for sec in time_seconds]
-        ax_errors.plot(time_seconds, error_counts, "-o", ms=3)
-    ax_errors.set_title("Errors per second")
-    ax_errors.set_xlabel("time (s)")
-    ax_errors.set_ylabel("errors / sec")
-
-    # Ensure unique output path and create directory if needed
-    final_out_path = get_incremented_path(out_path)
-    out_dir = os.path.dirname(final_out_path)
-    if out_dir:
-        os.makedirs(out_dir, exist_ok=True)
-
-    plt.tight_layout(rect=[0, 0, 1, 0.96])
-    plt.savefig(final_out_path, dpi=120)
-    print(f"Saved report to: {final_out_path}")
-
-    # Per-worker latency boxplot (top 12 by volume)
-    groups = {}
-    for r in oks:
-        host = urlparse(r.worker_url).netloc
-        groups.setdefault(host, []).append(r.total_ms)
-    items = sorted(groups.items(), key=lambda kv: len(kv[1]), reverse=True)[:12]
-    if items:
-        labels, data = zip(*items)
-        fig2, axb = plt.subplots(1, 1, figsize=(12, 5))
-        axb.boxplot(data, showfliers=False)
-        axb.set_xticklabels(labels, rotation=45, ha="right", fontsize=8)
-        axb.set_title("Per-worker latency (ms)")
-        axb.set_ylabel("ms")
-        plt.tight_layout()
-        extra_out = get_incremented_path(os.path.splitext(out_path)[0] + "-workers.png")
-        plt.savefig(extra_out, dpi=120)
-        fig2.tight_layout()
-        fig2.savefig(extra_out, dpi=120)
-        print(f"Saved worker latency plot to: {extra_out}")
+WORKER_ENDPOINT = "/v1/completions"

 if __name__ == "__main__":
    # Check if MODEL_NAME environment variable is set
@@ -403,32 +16,13 @@ if __name__ == "__main__":
        help="Model to use for completions request (required if MODEL_NAME env var not set)",
    )

-    # Parse known args to get model early, before adding load args
+    # Parse known args to get model early, before test_load_cmd adds its args
    known_args, _ = test_args.parse_known_args()
+
+    # Set environment variable if model was provided
    if hasattr(known_args, "model") and known_args.model:
        os.environ["MODEL_NAME"] = known_args.model
        print(f"Set MODEL_NAME environment variable to: {known_args.model}")

-    # Load test args
-    test_args.add_argument("-n", dest="num_requests", type=int, required=True, help="total number of requests")
-    test_args.add_argument("-rps", dest="requests_per_second", type=float, required=True, help="requests per second")
-    test_args.add_argument("--out", dest="out_path", type=str, default="load_test_report.png", help="path to save the report image")
-    args = test_args.parse_args()
-
-    server_url = {
-        "prod": "https://run.vast.ai",
-        "alpha": "https://run-alpha.vast.ai",
-        "candidate": "https://run-candidate.vast.ai",
-        "local": "http://localhost:8080"
-    }.get(args.instance, "http://localhost:8080")
-
-    run_load_with_metrics(
-        num_requests=args.num_requests,
-        requests_per_second=args.requests_per_second,
-        endpoint_group_name=args.endpoint_group_name,
-        account_api_key=args.api_key,
-        server_url=server_url,
-        worker_endpoint=WORKER_ENDPOINT,
-        instance=args.instance,
-        out_path=args.out_path,
-    )
+    # Now call test_load_cmd normally - it will add its own args and re-parse
+    test_load_cmd(CompletionsData, WORKER_ENDPOINT, arg_parser=test_args)