From 648cee226e0a9c21b0bd3efff1d9b6fc3d18b03e Mon Sep 17 00:00:00 2001 From: Duncan McQueen Date: Fri, 9 Oct 2026 06:45:56 -0500 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20strata=20proxy=20engine=20=E2=80=94?= =?UTF-8?q?=20front=20a=20remote=20Strata=20server=20as=20an=20OpenArc=20m?= =?UTF-8?q?odel?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PLAN.md task 4.2. Registers engine "strata" (model_type llm): model_path is the Strata server's base URL (or runtime_config.endpoint). StrataProxyLLM implements the engine contract over the remote OpenAI-compatible API: SSE streaming with reasoning_content passthrough, non-streaming, per-request cancel (closes the upstream stream), client-measured metrics marked proxy: true. Registry/add/CLI handle URL model paths (skip IR validation, skip CACHE_DIR injection, don't resolve URLs against the config dir). openarc add --mn flash-next --m http://localhost:8080 --en strata --mt llm --d strata 17 new unit tests (httpx.MockTransport); 316 pass, 2 pre-existing kokoro failures unchanged. --- docs/configure.md | 19 +- src/cli/groups/add.py | 18 +- src/cli/modules/server_config.py | 6 +- src/engine/strata_proxy.py | 404 +++++++++++++++++++++++++++ src/server/model_registry.py | 1 + src/server/schemas/registration.py | 10 +- tests/unit/test_strata_proxy_unit.py | 346 +++++++++++++++++++++++ 7 files changed, 793 insertions(+), 11 deletions(-) create mode 100644 src/engine/strata_proxy.py create mode 100644 tests/unit/test_strata_proxy_unit.py diff --git a/docs/configure.md b/docs/configure.md index c83d89ae..581cd884 100644 --- a/docs/configure.md +++ b/docs/configure.md @@ -6,7 +6,7 @@ OpenArc uses a YAML based configuration system! Models are added to `config.yaml There are a few different blocks in a models config that accept different parameters. Some will be common across implementations while others are specific. `openarc add` writes to these blocks; `openarc add --help` shows one help panel per key. ### load_config -- engine: ovgenai, openvino, optimum +- engine: ovgenai, openvino, optimum, strata - model_type: llm, vlm, whisper, kokoro, emb, rerank, qwen3_tts_* - device: device for this model - tool_call_parser: gemma4, qwen35, hermes @@ -163,6 +163,23 @@ openarc add \ --runtime-config '{"PERFORMANCE_HINT": "LATENCY"}' ``` +### Strata (remote proxy) + +`engine=strata` does not load a local model: it proxies requests to a running [Strata](https://github.com/Niko1221/Strata) server's OpenAI-compatible API, which lets OpenArc front models OpenVINO cannot run (GGUF MoE models, for example). `--model-path` carries the endpoint base URL instead of an IR directory, and `--device` is a placeholder. The served model id is discovered from the server's `GET /v1/models` at load time. Alternatively, set the URL as `endpoint` in `--runtime-config` and use any placeholder for `--model-path`. + +Metrics for strata models (`ttft`, throughputs, token counts) are measured client-side from the upstream usage event; they carry a `proxy: true` marker. + +``` +openarc add \ + --model-name flash-next \ + --model-path http://localhost:8080 \ + --engine strata \ + --model-type llm \ + --device strata \ + --temperature 0.7 \ + --max-tokens 4096 +``` + ### Qwen3-TTS All 3 flavors of qwen3-tts are supported, with a rich set of configuration options. PRs to improve performance or other enhancements are welcome diff --git a/src/cli/groups/add.py b/src/cli/groups/add.py index fa45c689..4052f826 100644 --- a/src/cli/groups/add.py +++ b/src/cli/groups/add.py @@ -64,11 +64,11 @@ def _fail(extra: str = "") -> NoReturn: help='Public facing name of the model.') @click.option('--model-path', '--m', required=True, - help='Path to OpenVINO IR converted model.') + help="Path to OpenVINO IR converted model. For engine=strata this is the endpoint base URL instead (e.g. http://localhost:8080).") @click.option('--engine', '--en', - type=click.Choice(['ovgenai', 'openvino', 'optimum']), + type=click.Choice(['ovgenai', 'openvino', 'optimum', 'strata']), required=True, - help='Engine used to load the model (ovgenai, openvino, optimum)') + help='Engine used to load the model (ovgenai, openvino, optimum, strata)') @click.option('--model-type', '--mt', type=click.Choice([ 'llm', 'vlm', 'whisper', 'qwen3_asr', 'kokoro', @@ -135,8 +135,13 @@ def add(ctx, model_path, model_name, engine, model_type, device, runtime_config, keeps using the contract's own default. """ - # Validate model path - if not validate_model_path(model_path): + # Validate model path. engine=strata carries an endpoint URL in model_path, + # not an on-disk IR directory, so the file check does not apply. + if engine == "strata": + if "://" not in model_path: + console.print(f"[red]engine=strata expects an endpoint URL in --model-path (e.g. http://localhost:8080), got: {model_path}[/red]") + ctx.exit(1) + elif not validate_model_path(model_path): console.print(f"[red]Model file check failed! {model_path} does not contain openvino model files OR your chosen path is malformed. Verify chosen path is correct and acquired model files match source on the hub, or the destination of converted model.[/red]") ctx.exit(1) @@ -157,7 +162,8 @@ def add(ctx, model_path, model_name, engine, model_type, device, runtime_config, # An explicit CACHE_DIR in --runtime-config wins. Resolved to an absolute # path here because runtime_config values are passed to OpenVINO verbatim, # unlike model_path they are not resolved against the config file. - if model_type in ("llm", "vlm"): + # engine=strata has no IR (model_path is a URL), so no cache applies. + if model_type in ("llm", "vlm") and engine != "strata": parsed_runtime_config.setdefault( "CACHE_DIR", str((Path(model_path) / "model_cache").resolve()) ) diff --git a/src/cli/modules/server_config.py b/src/cli/modules/server_config.py index 17468ed6..1bf00876 100644 --- a/src/cli/modules/server_config.py +++ b/src/cli/modules/server_config.py @@ -263,11 +263,13 @@ def get_all_models(self) -> Dict[str, Dict[str, Any]]: def _resolve_model_paths(self, model_config: Dict[str, Any]) -> Dict[str, Any]: """Return a copy of model_config with relative model_path, draft_model_path, - and cache_dir made absolute by joining them onto the config file's directory.""" + and cache_dir made absolute by joining them onto the config file's directory. + + URL values (e.g. a strata engine's endpoint in model_path) are left alone.""" resolved = dict(model_config) path = resolved.get("model_path") - if path and not Path(path).is_absolute(): + if path and "://" not in path and not Path(path).is_absolute(): resolved["model_path"] = str((self.config_file.parent / path).resolve()) draft_model_path = resolved.get("draft_model_path") diff --git a/src/engine/strata_proxy.py b/src/engine/strata_proxy.py new file mode 100644 index 00000000..1c40e6b4 --- /dev/null +++ b/src/engine/strata_proxy.py @@ -0,0 +1,404 @@ +"""Strata proxy engine: front a remote Strata server's OpenAI-compatible API. + +A "model" registered with engine=strata is not loaded locally; it is an HTTP +endpoint (https://github.com/Niko1221/Strata) serving /v1/chat/completions for +models OpenVINO cannot run (e.g. GGUF MoE models such as Qwen3.8-Flash-Next). + +Configuration +------------- +``load_config.model_path`` carries the endpoint base URL (e.g. +``http://localhost:8080``) -- there is no on-disk IR, so the CLI skips the IR +file check for this engine and config loading never resolves the value as a +path. Alternatively ``runtime_config.endpoint`` overrides model_path when set +(useful when a placeholder model_path reads better in config.yaml). + +``load_config.device`` is required by the schema but unused by this engine; +any placeholder string (e.g. "strata") is fine. + +Registration +------------ +``MODEL_CLASS_REGISTRY[(EngineType.STRATA, ModelType.LLM)]`` maps here. The +class subclasses OVGenAI_LLM purely so WorkerRegistry's isinstance routing +spawns an LLM queue worker without any worker changes; every load/generate +method is overridden and no OpenVINO pipeline is created. + +Request mapping (OVGenAI_GenConfig -> OpenAI chat/completions body) +------------------------------------------------------------------- +messages pass through verbatim (the remote server applies the chat template); +a raw ``prompt`` becomes a single user message. temperature (0 = greedy +upstream), top_p, top_k, max_tokens, seed, frequency_penalty, +presence_penalty, repetition_penalty, tools and chat_template_kwargs are +forwarded. Streaming requests ask for ``stream_options.include_usage`` so the +final SSE event carries token counts. + +Metrics +------- +OpenArc's collect_metrics shape, but measured client-side: TTFT is the wall +time from request send to the first content/reasoning chunk, throughputs are +derived from the upstream usage object and elapsed time. The 'proxy' key marks +the dict as client-measured rather than engine-reported. + +Cancellation +------------ +Per-request the active streaming httpx.Response is tracked by request_id; +cancel() closes it, which the remote server observes as a client disconnect +and aborts generation. +""" + +import asyncio +import json +import logging +import time +from typing import Any, AsyncIterator, Dict, List, Optional, Union + +import httpx + +from src.engine.ov_genai.llm import OVGenAI_LLM +from src.server.model_registry import ModelRegistry +from src.server.schemas.modeling.contract_ovgenai_llm_and_vlm import OVGenAI_GenConfig +from src.server.schemas.registration import ModelLoadConfig + +logger = logging.getLogger(__name__) + +# Long generations must not trip a read timeout; connect stays bounded so a +# dead endpoint fails fast instead of hanging a queue worker. +_CLIENT_TIMEOUT = httpx.Timeout(connect=10.0, read=None, write=30.0, pool=10.0) +_CLIENT_LIMITS = httpx.Limits(max_connections=20, max_keepalive_connections=10) +_PROBE_TIMEOUT = 10.0 + + +class StrataProxyLLM(OVGenAI_LLM): + """Engine proxying generation to a remote Strata (OpenAI-compatible) server.""" + + def __init__(self, load_config: ModelLoadConfig, transport: Optional[httpx.BaseTransport] = None): + super().__init__(load_config) + # transport: test hook (httpx.MockTransport works for sync and async clients) + self._transport = transport + self.endpoint: str = "" + self.served_model_id: Optional[str] = None + self._client: Optional[httpx.AsyncClient] = None + self._load_time_s: float = 0.0 + # request_id -> active streaming response (for cancel) + self._active_streams: Dict[str, httpx.Response] = {} + self._cancelled: set = set() + + # ------------------------------------------------------------------ load + + @staticmethod + def _resolve_endpoint(loader: ModelLoadConfig) -> str: + """Endpoint base URL from runtime_config.endpoint, else model_path.""" + endpoint = (loader.runtime_config or {}).get("endpoint") or loader.model_path + if not isinstance(endpoint, str) or "://" not in endpoint: + raise ValueError( + f"[{loader.model_name}] engine=strata needs the Strata endpoint URL in " + f"model_path (e.g. http://localhost:8080) or runtime_config.endpoint; " + f"got {endpoint!r}" + ) + endpoint = endpoint.rstrip("/") + if endpoint.endswith("/v1"): + endpoint = endpoint[: -len("/v1")] + return endpoint + + def load_model(self, loader: ModelLoadConfig) -> None: + """Verify the endpoint is reachable and capture the served model id. + + Runs synchronously (the registry calls load_model via to_thread), so the + reachability probe uses a short-lived sync client; the streaming client + is only constructed here and first used inside the event loop. + """ + self.endpoint = self._resolve_endpoint(loader) + logger.info(f"{loader.model_name} probing Strata endpoint {self.endpoint} ...") + + start = time.perf_counter() + try: + with httpx.Client(timeout=_PROBE_TIMEOUT, transport=self._transport) as probe: + resp = probe.get(f"{self.endpoint}/v1/models") + except httpx.HTTPError as exc: + raise RuntimeError( + f"[{loader.model_name}] Strata endpoint {self.endpoint} unreachable: {exc}" + ) from exc + + if resp.status_code != 200: + raise RuntimeError( + f"[{loader.model_name}] Strata endpoint {self.endpoint} answered " + f"GET /v1/models with status {resp.status_code}: {resp.text[:500]}" + ) + + try: + data = resp.json().get("data") or [] + except json.JSONDecodeError as exc: + raise RuntimeError( + f"[{loader.model_name}] Strata endpoint {self.endpoint} returned " + f"non-JSON from /v1/models" + ) from exc + if not data: + raise RuntimeError( + f"[{loader.model_name}] Strata endpoint {self.endpoint} is reachable " + f"but serves no models" + ) + self.served_model_id = data[0].get("id") + self._load_time_s = time.perf_counter() - start + + self._client = httpx.AsyncClient( + timeout=_CLIENT_TIMEOUT, + limits=_CLIENT_LIMITS, + transport=self._transport, + ) + logger.info( + f"{loader.model_name} loaded successfully " + f"(proxy to {self.endpoint}, serving model id '{self.served_model_id}')" + ) + + # -------------------------------------------------------------- generate + + def generate_type(self, gen_config: OVGenAI_GenConfig): + """ + Route to streaming or non-streaming generation. + + Returns: + - Non-streaming: async iterator yielding [metrics: dict, new_text: str] + - Streaming: async iterator yielding str chunks / {"chat_delta": [...]} + dicts, then a terminal metrics dict + """ + if gen_config.stream: + return self.generate_stream(gen_config) + return self.generate_text(gen_config) + + def _build_request_body(self, gen_config: OVGenAI_GenConfig, stream: bool) -> Dict[str, Any]: + if gen_config.input_ids: + raise ValueError( + "engine=strata does not accept pre-encoded input_ids " + "(/openarc/bench is unsupported); send a prompt or messages" + ) + if gen_config.messages: + messages = gen_config.messages + elif gen_config.prompt: + messages = [{"role": "user", "content": gen_config.prompt}] + else: + raise ValueError("engine=strata request needs messages or a prompt") + + body: Dict[str, Any] = { + "model": self.served_model_id, + "messages": messages, + "temperature": gen_config.temperature, # 0 means greedy upstream + "top_p": gen_config.top_p, + "top_k": gen_config.top_k, + "max_tokens": gen_config.max_tokens, + "stream": stream, + } + if gen_config.seed is not None: + body["seed"] = gen_config.seed + if gen_config.repetition_penalty != 1.0: + body["repetition_penalty"] = gen_config.repetition_penalty + if gen_config.frequency_penalty is not None: + body["frequency_penalty"] = gen_config.frequency_penalty + if gen_config.presence_penalty is not None: + body["presence_penalty"] = gen_config.presence_penalty + if gen_config.tools: + body["tools"] = gen_config.tools + if gen_config.chat_template_kwargs: + body["chat_template_kwargs"] = gen_config.chat_template_kwargs + if stream: + # Ask the server for a final usage event so token counts are real. + body["stream_options"] = {"include_usage": True} + return body + + async def _post_chat(self, body: Dict[str, Any]) -> httpx.Response: + assert self._client is not None, "load_model must run before generate_type" + url = f"{self.endpoint}/v1/chat/completions" + try: + resp = await self._client.post(url, json=body) + except httpx.HTTPError as exc: + raise RuntimeError( + f"[{self.load_config.model_name}] Strata request failed: {exc}" + ) from exc + if resp.status_code != 200: + raise RuntimeError( + f"[{self.load_config.model_name}] Strata answered status " + f"{resp.status_code}: {resp.text[:500]}" + ) + return resp + + async def generate_text(self, gen_config: OVGenAI_GenConfig) -> AsyncIterator[Union[Dict[str, Any], str]]: + """Non-streaming generation. Yields metrics (dict), then new_text (str).""" + body = self._build_request_body(gen_config, stream=False) + start = time.perf_counter() + resp = await self._post_chat(body) + elapsed = time.perf_counter() - start + + payload = resp.json() + choice = (payload.get("choices") or [{}])[0] + message = choice.get("message") or {} + text = message.get("content") or "" + + yield self.collect_metrics( + gen_config, + usage=payload.get("usage"), + ttft_s=elapsed, + elapsed_s=elapsed, + ) + yield text + + async def generate_stream(self, gen_config: OVGenAI_GenConfig) -> AsyncIterator[Union[str, Dict[str, Any]]]: + """Streaming generation. + + Yields str content chunks as SSE deltas arrive, reasoning deltas as + {"chat_delta": [{"reasoning_content": ...}]} (the item protocol + InferWorker.infer_llm already forwards), then a terminal metrics dict. + """ + body = self._build_request_body(gen_config, stream=True) + request_id = gen_config.request_id + assert self._client is not None, "load_model must run before generate_type" + url = f"{self.endpoint}/v1/chat/completions" + + start = time.perf_counter() + ttft_s: Optional[float] = None + usage: Optional[Dict[str, Any]] = None + + try: + async with self._client.stream("POST", url, json=body) as resp: + if resp.status_code != 200: + error_text = (await resp.aread()).decode(errors="replace") + raise RuntimeError( + f"[{self.load_config.model_name}] Strata answered status " + f"{resp.status_code}: {error_text[:500]}" + ) + if request_id: + self._active_streams[request_id] = resp + + async for line in resp.aiter_lines(): + line = line.strip() + if not line.startswith("data:"): + continue + data = line[len("data:"):].strip() + if data == "[DONE]": + break + try: + event = json.loads(data) + except json.JSONDecodeError: + logger.debug(f"[{self.load_config.model_name}] undecodable SSE line: {data[:200]}") + continue + + if event.get("usage"): + usage = event["usage"] + + choices = event.get("choices") or [] + if not choices: + continue + delta = choices[0].get("delta") or {} + reasoning = delta.get("reasoning_content") + content = delta.get("content") + + if (reasoning or content) and ttft_s is None: + ttft_s = time.perf_counter() - start + if reasoning: + yield {"chat_delta": [{"reasoning_content": reasoning}]} + if content: + yield content + except httpx.StreamError as exc: + if request_id and request_id in self._cancelled: + # cancel() closed the stream; end quietly, no metrics. + logger.info( + f"[{self.load_config.model_name}] request {request_id} stream closed by cancel" + ) + return + raise RuntimeError( + f"[{self.load_config.model_name}] Strata stream interrupted: {exc}" + ) from exc + except httpx.HTTPError as exc: + raise RuntimeError( + f"[{self.load_config.model_name}] Strata request failed: {exc}" + ) from exc + finally: + if request_id: + self._active_streams.pop(request_id, None) + self._cancelled.discard(request_id) + + elapsed = time.perf_counter() - start + yield self.collect_metrics( + gen_config, + usage=usage, + ttft_s=ttft_s if ttft_s is not None else elapsed, + elapsed_s=elapsed, + ) + + # --------------------------------------------------------------- metrics + + def collect_metrics( + self, + gen_config: OVGenAI_GenConfig, + usage: Optional[Dict[str, Any]], + ttft_s: float, + elapsed_s: float, + ) -> Dict[str, Any]: + """OpenArc metrics shape, measured client-side from the proxy. + + 'proxy' marks every value as client-measured (httpx timers plus the + upstream usage object), not engine-reported PerfMetrics. + """ + usage = usage or {} + input_tokens = int(usage.get("prompt_tokens") or 0) + new_tokens = int(usage.get("completion_tokens") or 0) + + decode_s = max(elapsed_s - ttft_s, 0.0) + # Non-streaming has no separate decode window (ttft == elapsed): fall + # back to total elapsed so tpot/throughput are effective values. + effective_decode_s = decode_s if decode_s > 0 else elapsed_s + prefill_throughput = round(input_tokens / ttft_s, 2) if ttft_s > 0 else 0 + tpot_ms = ( + round(effective_decode_s / max(new_tokens - 1, 1) * 1000, 5) + if new_tokens > 0 and effective_decode_s > 0 + else 0 + ) + decode_throughput = ( + round(new_tokens / effective_decode_s, 5) + if new_tokens > 0 and effective_decode_s > 0 + else 0 + ) + + return { + 'load_time (s)': round(self._load_time_s, 2), + 'ttft (s)': round(ttft_s, 2), + 'tpot (ms)': tpot_ms, + 'prefill_throughput (tokens/s)': prefill_throughput, + 'decode_throughput (tokens/s)': decode_throughput, + 'decode_duration (s)': round(decode_s, 5), + 'input_token': input_tokens, + 'new_token': new_tokens, + 'total_token': input_tokens + new_tokens, + 'stream': gen_config.stream, + 'proxy': True, + 'endpoint': self.endpoint, + } + + # --------------------------------------------------------------- control + + async def cancel(self, request_id: str) -> bool: + """Cancel an ongoing streaming generation by request_id. + + Closes the active httpx stream; the remote server sees a client + disconnect and aborts the request. + """ + resp = self._active_streams.get(request_id) + if resp is None: + return False + self._cancelled.add(request_id) + try: + await resp.aclose() + except Exception as exc: + # The read side may hold the stream mid-chunk; it will still see + # the close (or the _cancelled mark) and wind down. + logger.debug(f"[{self.load_config.model_name}] stream close raised: {exc}") + logger.info(f"[{self.load_config.model_name}] Cancellation triggered for request {request_id}") + return True + + async def unload_model(self, registry: ModelRegistry, model_name: str) -> bool: + """Unregister the model and close the shared HTTP client.""" + removed = await registry.register_unload(model_name) + + if self._client is not None: + await self._client.aclose() + self._client = None + + logging.info(f"[{self.load_config.model_name}] unloaded successfully") + return removed diff --git a/src/server/model_registry.py b/src/server/model_registry.py index fe8b0ffc..c942534f 100644 --- a/src/server/model_registry.py +++ b/src/server/model_registry.py @@ -345,6 +345,7 @@ async def readiness(self) -> dict: (EngineType.OPENVINO, ModelType.QWEN3_TTS_VOICE_CLONE): "src.engine.openvino.qwen3_tts.qwen3_tts.OVQwen3TTS", (EngineType.OV_OPTIMUM, ModelType.EMB): "src.engine.optimum.optimum_emb.Optimum_EMB", (EngineType.OV_OPTIMUM, ModelType.RERANK): "src.engine.optimum.optimum_rr.Optimum_RR", + (EngineType.STRATA, ModelType.LLM): "src.engine.strata_proxy.StrataProxyLLM", } async def create_model_instance(load_config: ModelLoadConfig) -> Any: diff --git a/src/server/schemas/registration.py b/src/server/schemas/registration.py index 09d97839..9fea7cfa 100644 --- a/src/server/schemas/registration.py +++ b/src/server/schemas/registration.py @@ -53,11 +53,14 @@ class EngineType(str, Enum): Options: - optimum: Optimum-Intel engine - - ovgenai: OpenVINO GenAI engine""" + - ovgenai: OpenVINO GenAI engine + - openvino: OpenVINO engine + - strata: proxy to a remote Strata server's OpenAI-compatible API""" OV_OPTIMUM = "optimum" OV_GENAI = "ovgenai" OPENVINO = "openvino" + STRATA = "strata" class ToolCallParser(str, Enum): @@ -85,7 +88,10 @@ class ModelLoadConfig(BaseModel): description=""" Top level path to directory containing OpenVINO IR converted model. - OpenArc does not support runtime conversion and cannot pull from HF.""") + OpenArc does not support runtime conversion and cannot pull from HF. + + For engine=strata this instead carries the endpoint base URL of the + remote Strata server (e.g. http://localhost:8080).""") model_name: str = Field( ..., description=""" diff --git a/tests/unit/test_strata_proxy_unit.py b/tests/unit/test_strata_proxy_unit.py new file mode 100644 index 00000000..a5b73c5b --- /dev/null +++ b/tests/unit/test_strata_proxy_unit.py @@ -0,0 +1,346 @@ +from importlib import import_module +import json +from unittest.mock import AsyncMock, MagicMock + +import httpx +import pytest # type: ignore[import] + +from src.engine.strata_proxy import StrataProxyLLM +from src.server.model_registry import MODEL_CLASS_REGISTRY +from src.server.schemas.modeling.contract_ovgenai_llm_and_vlm import OVGenAI_GenConfig +from src.server.schemas.registration import EngineType, ModelLoadConfig, ModelType + + +ENDPOINT = "http://strata.test:8080" +SERVED_ID = "qwen3.8-flash-next-iq3_s" + + +def sse(events) -> bytes: + lines = [] + for event in events: + if event == "[DONE]": + lines.append("data: [DONE]") + else: + lines.append(f"data: {json.dumps(event)}") + return ("\n\n".join(lines) + "\n\n").encode() + + +def delta(**fields): + return {"choices": [{"index": 0, "delta": fields, "finish_reason": None}]} + + +USAGE = {"prompt_tokens": 12, "completion_tokens": 5, "total_tokens": 17} + + +def make_transport(handler): + return httpx.MockTransport(handler) + + +@pytest.fixture +def load_config() -> ModelLoadConfig: + return ModelLoadConfig( + model_path=ENDPOINT, + model_name="strata-proxy", + model_type=ModelType.LLM, + engine=EngineType.STRATA, + device="strata", + ) + + +def make_engine(load_config: ModelLoadConfig, handler) -> StrataProxyLLM: + engine = StrataProxyLLM(load_config, transport=make_transport(handler)) + engine.load_model(load_config) + return engine + + +def ok_models_handler(request: httpx.Request) -> httpx.Response: + assert request.url.path == "/v1/models" + return httpx.Response(200, json={"object": "list", "data": [{"id": SERVED_ID}]}) + + +# --------------------------------------------------------------------- load + + +def test_load_model_captures_served_model_id(load_config: ModelLoadConfig) -> None: + engine = make_engine(load_config, ok_models_handler) + + assert engine.endpoint == ENDPOINT + assert engine.served_model_id == SERVED_ID + assert engine._client is not None + + +def test_load_model_normalizes_trailing_slash_and_v1(load_config: ModelLoadConfig) -> None: + for raw in (ENDPOINT + "/", ENDPOINT + "/v1"): + loader = load_config.model_copy(update={"model_path": raw}) + engine = make_engine(loader, ok_models_handler) + assert engine.endpoint == ENDPOINT + + +def test_load_model_endpoint_from_runtime_config(load_config: ModelLoadConfig) -> None: + loader = load_config.model_copy( + update={"model_path": "placeholder", "runtime_config": {"endpoint": ENDPOINT}} + ) + engine = make_engine(loader, ok_models_handler) + assert engine.endpoint == ENDPOINT + + +def test_load_model_rejects_non_url(load_config: ModelLoadConfig) -> None: + loader = load_config.model_copy(update={"model_path": "/models/some-ir"}) + with pytest.raises(ValueError, match="endpoint URL"): + StrataProxyLLM(loader).load_model(loader) + + +def test_load_model_unreachable_raises_clear_error(load_config: ModelLoadConfig) -> None: + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("connection refused", request=request) + + with pytest.raises(RuntimeError, match="unreachable"): + make_engine(load_config, handler) + + +def test_load_model_non_200_models_raises(load_config: ModelLoadConfig) -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(500, text="boom") + + with pytest.raises(RuntimeError, match="status 500"): + make_engine(load_config, handler) + + +def test_load_model_no_models_served_raises(load_config: ModelLoadConfig) -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"object": "list", "data": []}) + + with pytest.raises(RuntimeError, match="no models"): + make_engine(load_config, handler) + + +# ---------------------------------------------------------------- streaming + + +def streaming_handler(captured: dict): + events = [ + delta(role="assistant"), + delta(reasoning_content="thinking..."), + delta(content="Hello"), + delta(content=" world"), + {"choices": [], "usage": USAGE}, + "[DONE]", + ] + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/v1/models": + return ok_models_handler(request) + captured["body"] = json.loads(request.content) + return httpx.Response( + 200, + content=sse(events), + headers={"content-type": "text/event-stream"}, + ) + + return handler + + +@pytest.mark.asyncio +async def test_generate_stream_yields_chunks_then_metrics(load_config: ModelLoadConfig) -> None: + captured: dict = {} + engine = make_engine(load_config, streaming_handler(captured)) + + gen_config = OVGenAI_GenConfig( + messages=[{"role": "user", "content": "hi"}], + stream=True, + request_id="req-stream", + temperature=0.0, + seed=42, + chat_template_kwargs={"enable_thinking": False}, + ) + + items = [item async for item in engine.generate_type(gen_config)] + + assert items[0] == {"chat_delta": [{"reasoning_content": "thinking..."}]} + assert items[1] == "Hello" + assert items[2] == " world" + + metrics = items[-1] + assert metrics["input_token"] == USAGE["prompt_tokens"] + assert metrics["new_token"] == USAGE["completion_tokens"] + assert metrics["total_token"] == USAGE["total_tokens"] + assert metrics["stream"] is True + assert metrics["proxy"] is True + assert metrics["endpoint"] == ENDPOINT + assert metrics["ttft (s)"] >= 0 + + body = captured["body"] + assert body["model"] == SERVED_ID + assert body["stream"] is True + assert body["stream_options"] == {"include_usage": True} + assert body["messages"] == [{"role": "user", "content": "hi"}] + assert body["temperature"] == 0.0 + assert body["seed"] == 42 + assert body["chat_template_kwargs"] == {"enable_thinking": False} + + +@pytest.mark.asyncio +async def test_generate_stream_prompt_becomes_user_message(load_config: ModelLoadConfig) -> None: + captured: dict = {} + engine = make_engine(load_config, streaming_handler(captured)) + + gen_config = OVGenAI_GenConfig(prompt="raw prompt", stream=True) + async for _ in engine.generate_type(gen_config): + pass + + assert captured["body"]["messages"] == [{"role": "user", "content": "raw prompt"}] + + +@pytest.mark.asyncio +async def test_generate_stream_upstream_error_raises(load_config: ModelLoadConfig) -> None: + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/v1/models": + return ok_models_handler(request) + return httpx.Response(500, text="engine exploded") + + engine = make_engine(load_config, handler) + gen_config = OVGenAI_GenConfig( + messages=[{"role": "user", "content": "hi"}], stream=True, request_id="req-err" + ) + + with pytest.raises(RuntimeError, match="status 500"): + async for _ in engine.generate_type(gen_config): + pass + + +@pytest.mark.asyncio +async def test_generate_stream_input_ids_rejected(load_config: ModelLoadConfig) -> None: + engine = make_engine(load_config, ok_models_handler) + gen_config = OVGenAI_GenConfig(input_ids=[1, 2, 3], stream=True) + + with pytest.raises(ValueError, match="input_ids"): + async for _ in engine.generate_type(gen_config): + pass + + +# ------------------------------------------------------------- non-streaming + + +@pytest.mark.asyncio +async def test_generate_text_yields_metrics_then_text(load_config: ModelLoadConfig) -> None: + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/v1/models": + return ok_models_handler(request) + captured["body"] = json.loads(request.content) + return httpx.Response(200, json={ + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": "full answer"}, + "finish_reason": "stop", + }], + "usage": USAGE, + }) + + engine = make_engine(load_config, handler) + gen_config = OVGenAI_GenConfig( + messages=[{"role": "user", "content": "hi"}], stream=False + ) + + items = [item async for item in engine.generate_type(gen_config)] + + assert len(items) == 2 + metrics, text = items + assert text == "full answer" + assert metrics["input_token"] == USAGE["prompt_tokens"] + assert metrics["new_token"] == USAGE["completion_tokens"] + assert metrics["stream"] is False + assert metrics["proxy"] is True + assert captured["body"]["stream"] is False + assert "stream_options" not in captured["body"] + + +# ------------------------------------------------------------------- cancel + + +@pytest.mark.asyncio +async def test_cancel_closes_stream_and_returns_true(load_config: ModelLoadConfig) -> None: + engine_holder: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/v1/models": + return ok_models_handler(request) + + async def body(): + yield sse([delta(content="Hello")]) + # A real server streams until the client disconnects; here the + # generator hangs until cancel() marks the request, then surfaces + # the same StreamClosed a closed httpx stream would raise. + engine = engine_holder["engine"] + while "req-cancel" not in engine._cancelled: + import asyncio + await asyncio.sleep(0.001) + raise httpx.StreamClosed() + + return httpx.Response( + 200, content=body(), headers={"content-type": "text/event-stream"} + ) + + engine = make_engine(load_config, handler) + engine_holder["engine"] = engine + + gen_config = OVGenAI_GenConfig( + messages=[{"role": "user", "content": "hi"}], stream=True, request_id="req-cancel" + ) + stream = engine.generate_type(gen_config) + + first = await stream.__anext__() + assert first == "Hello" + assert "req-cancel" in engine._active_streams + + assert await engine.cancel("req-cancel") is True + + remaining = [item async for item in stream] + assert remaining == [] # cancelled stream ends quietly, without metrics + assert "req-cancel" not in engine._active_streams + + +@pytest.mark.asyncio +async def test_cancel_unknown_request_id_returns_false(load_config: ModelLoadConfig) -> None: + engine = make_engine(load_config, ok_models_handler) + assert await engine.cancel("no-such-request") is False + + +@pytest.mark.asyncio +async def test_cancel_calls_response_aclose(load_config: ModelLoadConfig) -> None: + engine = make_engine(load_config, ok_models_handler) + fake_response = MagicMock() + fake_response.aclose = AsyncMock() + engine._active_streams["req-x"] = fake_response + + assert await engine.cancel("req-x") is True + fake_response.aclose.assert_awaited_once() + assert "req-x" in engine._cancelled + + +# ------------------------------------------------------------------- unload + + +@pytest.mark.asyncio +async def test_unload_model_closes_client(load_config: ModelLoadConfig) -> None: + engine = make_engine(load_config, ok_models_handler) + + registry = MagicMock() + registry.register_unload = AsyncMock(return_value=True) + + result = await engine.unload_model(registry, "strata-proxy") + + assert result is True + assert engine._client is None + registry.register_unload.assert_called_once_with("strata-proxy") + + +# ----------------------------------------------------------------- registry + + +def test_model_class_registry_resolves_strata_llm() -> None: + class_path = MODEL_CLASS_REGISTRY[(EngineType.STRATA, ModelType.LLM)] + module_path, class_name = class_path.rsplit(".", 1) + resolved = getattr(import_module(module_path), class_name) + assert resolved is StrataProxyLLM From 5f10676feedbc4a7f2d1e11c8feccc424e65f081 Mon Sep 17 00:00:00 2001 From: Duncan McQueen Date: Fri, 9 Oct 2026 07:11:50 -0500 Subject: [PATCH 2/3] fix(strata proxy): client disconnect must not unload the model httpx transport errors (ReadError) and stream-state errors (StreamClosed) are disjoint hierarchies -- StreamError is a RuntimeError, not an HTTPError. A cancel() that closed the upstream stream surfaced as ReadError in the generic HTTPError handler, raised RuntimeError, and the worker registry unloaded the model. Catch both with the cancelled-check. Live-tested against a real Strata server: model stays loaded and serving after a mid-stream client disconnect. Adds regression tests for both error types. --- src/engine/strata_proxy.py | 13 +++--- tests/unit/test_strata_proxy_unit.py | 62 ++++++++++++++++++++++++++++ 2 files changed, 69 insertions(+), 6 deletions(-) diff --git a/src/engine/strata_proxy.py b/src/engine/strata_proxy.py index 1c40e6b4..511bfbdc 100644 --- a/src/engine/strata_proxy.py +++ b/src/engine/strata_proxy.py @@ -295,17 +295,18 @@ async def generate_stream(self, gen_config: OVGenAI_GenConfig) -> AsyncIterator[ yield {"chat_delta": [{"reasoning_content": reasoning}]} if content: yield content - except httpx.StreamError as exc: + except (httpx.HTTPError, httpx.StreamError) as exc: + # Two disjoint hierarchies: transport errors (ReadError etc.) are + # HTTPError; stream-state errors (StreamClosed) are RuntimeErrors. if request_id and request_id in self._cancelled: - # cancel() closed the stream; end quietly, no metrics. + # cancel() closed the stream; the parked read surfaces as a + # transport or stream-state error depending on where it was + # blocked. A proxy must never let a client disconnect look + # like an inference failure: that unloads the registry entry. logger.info( f"[{self.load_config.model_name}] request {request_id} stream closed by cancel" ) return - raise RuntimeError( - f"[{self.load_config.model_name}] Strata stream interrupted: {exc}" - ) from exc - except httpx.HTTPError as exc: raise RuntimeError( f"[{self.load_config.model_name}] Strata request failed: {exc}" ) from exc diff --git a/tests/unit/test_strata_proxy_unit.py b/tests/unit/test_strata_proxy_unit.py index a5b73c5b..463b775f 100644 --- a/tests/unit/test_strata_proxy_unit.py +++ b/tests/unit/test_strata_proxy_unit.py @@ -307,6 +307,68 @@ async def test_cancel_unknown_request_id_returns_false(load_config: ModelLoadCon assert await engine.cancel("no-such-request") is False +@pytest.mark.asyncio +async def test_cancel_read_error_ends_quietly(load_config: ModelLoadConfig) -> None: + """Regression: a cancelled stream's parked read raises httpx.ReadError + (a transport error, NOT a StreamError). It must end quietly -- a RuntimeError + here makes the worker registry unload the model on a mere client disconnect.""" + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/v1/models": + return ok_models_handler(request) + + async def body(): + yield sse([delta(content="Hello")]) + raise httpx.ReadError("connection closed") + + return httpx.Response( + 200, content=body(), headers={"content-type": "text/event-stream"} + ) + + engine = make_engine(load_config, handler) + + gen_config = OVGenAI_GenConfig( + messages=[{"role": "user", "content": "hi"}], stream=True, request_id="req-read-err" + ) + stream = engine.generate_type(gen_config) + first = await stream.__anext__() + assert first == "Hello" + + # Mark as cancelled (what cancel() does) before the ReadError surfaces. + engine._cancelled.add("req-read-err") + + remaining = [item async for item in stream] + assert remaining == [] # no RuntimeError, no metrics + assert "req-read-err" not in engine._active_streams + + +@pytest.mark.asyncio +async def test_uncancelled_read_error_raises(load_config: ModelLoadConfig) -> None: + """A ReadError on a request nobody cancelled is a genuine upstream failure.""" + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/v1/models": + return ok_models_handler(request) + + async def body(): + yield sse([delta(content="Hello")]) + raise httpx.ReadError("connection closed") + + return httpx.Response( + 200, content=body(), headers={"content-type": "text/event-stream"} + ) + + engine = make_engine(load_config, handler) + gen_config = OVGenAI_GenConfig( + messages=[{"role": "user", "content": "hi"}], stream=True, request_id="req-real-err" + ) + stream = engine.generate_type(gen_config) + assert await stream.__anext__() == "Hello" + with pytest.raises(RuntimeError, match="Strata request failed"): + async for _ in stream: + pass + + @pytest.mark.asyncio async def test_cancel_calls_response_aclose(load_config: ModelLoadConfig) -> None: engine = make_engine(load_config, ok_models_handler) From c2e4fbc4f8f54e9762d0778bdde5296d30ff9fd8 Mon Sep 17 00:00:00 2001 From: Duncan McQueen Date: Fri, 9 Oct 2026 07:22:28 -0500 Subject: [PATCH 3/3] docs(strata proxy): document OpenAI-protocol-only decoupling from Strata The proxy speaks only the stable OpenAI wire protocol (models list, chat completions, SSE, usage) and assumes nothing about Strata's internals or version, so it tracks no Strata release cadence and works with any OpenAI-compatible server. --- docs/configure.md | 2 ++ src/engine/strata_proxy.py | 9 +++++++++ 2 files changed, 11 insertions(+) diff --git a/docs/configure.md b/docs/configure.md index 581cd884..98e4b0bd 100644 --- a/docs/configure.md +++ b/docs/configure.md @@ -167,6 +167,8 @@ openarc add \ `engine=strata` does not load a local model: it proxies requests to a running [Strata](https://github.com/Niko1221/Strata) server's OpenAI-compatible API, which lets OpenArc front models OpenVINO cannot run (GGUF MoE models, for example). `--model-path` carries the endpoint base URL instead of an IR directory, and `--device` is a placeholder. The served model id is discovered from the server's `GET /v1/models` at load time. Alternatively, set the URL as `endpoint` in `--runtime-config` and use any placeholder for `--model-path`. +The proxy deliberately speaks only the stable OpenAI wire protocol (`GET /v1/models`, `POST /v1/chat/completions`, SSE deltas, `usage` events) and nothing Strata-specific, so it is decoupled from Strata's release cadence — and works with any other OpenAI-compatible server that supports those endpoints. All proxy logic lives in the single module `src/engine/strata_proxy.py`. + Metrics for strata models (`ttft`, throughputs, token counts) are measured client-side from the upstream usage event; they carry a `proxy: true` marker. ``` diff --git a/src/engine/strata_proxy.py b/src/engine/strata_proxy.py index 511bfbdc..3dd77cbf 100644 --- a/src/engine/strata_proxy.py +++ b/src/engine/strata_proxy.py @@ -4,6 +4,15 @@ endpoint (https://github.com/Niko1221/Strata) serving /v1/chat/completions for models OpenVINO cannot run (e.g. GGUF MoE models such as Qwen3.8-Flash-Next). +Decoupling +---------- +Strata moves fast, so by design this module only speaks the stable OpenAI wire +protocol (GET /v1/models, POST /v1/chat/completions, SSE deltas, usage events) +and assumes nothing about Strata's internals or version. Any other +OpenAI-compatible server exposing those endpoints works too. Everything +Strata-related lives in this one file; the rest of OpenArc only sees the +engine registry entry and the "strata" EngineType. + Configuration ------------- ``load_config.model_path`` carries the endpoint base URL (e.g.