fix(ai): bind PixelRAG VLM worker to selected route

services/pixelrag_vlm_replay_worker_service.py

@@ -16,7 +16,15 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Mapping
 
-from services.ollama_service import OllamaService, get_host_label, get_provider_tag
+import requests
+
+from services.ollama_service import (
+    OllamaResponse,
+    OllamaService,
+    get_host_label,
+    get_provider_tag,
+    is_approved_ollama_host,
+)
 from services.pixelrag_crawler_integration_service import (
     DEFAULT_ARTIFACT_MAX_AGE_HOURS,
     DEFAULT_ARTIFACT_ROOT,
@@ -223,6 +231,78 @@ def _validate_model_payload(
     }
 
 
+def _generate_exact_host(
+    prompt: str,
+    *,
+    host: str,
+    model: str,
+    temperature: float,
+    timeout: int,
+    options: Mapping[str, Any] | None,
+    images: list[str],
+) -> OllamaResponse:
+    """Call the route-readiness selected host without fallback or model downgrade."""
+    clean_host = str(host or "").rstrip("/")
+    if not is_approved_ollama_host(clean_host):
+        return OllamaResponse(
+            success=False,
+            content="",
+            model=model,
+            error=f"unapproved_pixelrag_vlm_candidate_host: {clean_host}",
+            host=clean_host or "unknown",
+        )
+    payload: dict[str, Any] = {
+        "model": model,
+        "prompt": prompt,
+        "stream": False,
+        "options": {"temperature": temperature},
+    }
+    if options:
+        payload["options"].update(dict(options))
+    if images:
+        payload["images"] = images
+    try:
+        response = requests.post(
+            f"{clean_host}/api/generate",
+            json=payload,
+            timeout=max(1, int(timeout or DEFAULT_TIMEOUT_SECONDS)),
+        )
+        if response.status_code != 200:
+            return OllamaResponse(
+                success=False,
+                content="",
+                model=model,
+                error=f"HTTP {response.status_code}: {response.text[:RAW_EXCERPT_LIMIT]}",
+                host=clean_host,
+            )
+        data = response.json()
+        return OllamaResponse(
+            success=True,
+            content=data.get("response", ""),
+            model=model,
+            total_duration=(data.get("total_duration", 0) or 0) / 1e9,
+            host=clean_host,
+            input_tokens=int(data.get("prompt_eval_count", 0) or 0),
+            output_tokens=int(data.get("eval_count", 0) or 0),
+        )
+    except requests.Timeout:
+        return OllamaResponse(
+            success=False,
+            content="",
+            model=model,
+            error=f"timeout ({max(1, int(timeout or DEFAULT_TIMEOUT_SECONDS))}s)",
+            host=clean_host,
+        )
+    except Exception as exc:
+        return OllamaResponse(
+            success=False,
+            content="",
+            model=model,
+            error=f"{type(exc).__name__}: {str(exc)[:RAW_EXCERPT_LIMIT]}",
+            host=clean_host,
+        )
+
+
 def _write_replay_receipt(
     *,
     output_root: Path,
@@ -324,6 +404,7 @@ def _execute_item(
     root: Path,
     output_root: Path,
     model: str,
+    route_host: str | None,
     timeout: int,
     tile_limit: int,
     write_receipt: bool,
@@ -335,6 +416,7 @@ def _execute_item(
         "source_receipt_path": item.get("source_receipt_path"),
         "worker_status": "executing",
         "model": model,
+        "route_candidate_host": str(route_host or ""),
         "tile_evidence": tile_evidence,
         "tile_image_count": len(images),
         "model_call_performed": bool(images),
@@ -348,14 +430,27 @@ def _execute_item(
         })
         return base
 
-    response = OllamaService(model=model).generate(
-        _prompt_for_item(item),
-        model=model,
-        temperature=0.1,
-        timeout=max(10, int(timeout or DEFAULT_TIMEOUT_SECONDS)),
-        options={"num_predict": 700, "num_ctx": 4096},
-        images=images,
-    )
+    prompt = _prompt_for_item(item)
+    options = {"num_predict": 700, "num_ctx": 4096}
+    if route_host:
+        response = _generate_exact_host(
+            prompt,
+            host=route_host,
+            model=model,
+            temperature=0.1,
+            timeout=max(10, int(timeout or DEFAULT_TIMEOUT_SECONDS)),
+            options=options,
+            images=images,
+        )
+    else:
+        response = OllamaService(model=model).generate(
+            prompt,
+            model=model,
+            temperature=0.1,
+            timeout=max(10, int(timeout or DEFAULT_TIMEOUT_SECONDS)),
+            options=options,
+            images=images,
+        )
     base.update({
         "host": response.host,
         "host_label": get_host_label(response.host or ""),
@@ -369,7 +464,11 @@ def _execute_item(
         base.update({
             "worker_status": "model_error",
             "model_error": str(response.error or "")[:RAW_EXCERPT_LIMIT],
-            "next_machine_action": "repair_ollama_vlm_runtime_or_model_route",
+            "next_machine_action": (
+                "repair_ollama_vlm_generate_runtime_or_proxy_timeout"
+                if route_host
+                else "repair_ollama_vlm_runtime_or_model_route"
+            ),
         })
         if write_receipt:
             base["receipt_path"] = _write_replay_receipt(
@@ -454,6 +553,7 @@ def run_pixelrag_ollama_vlm_replay_worker(
     item_limit = max(1, min(int(limit or DEFAULT_LIMIT), 250))
     tiles = max(1, min(int(tile_limit or DEFAULT_TILE_LIMIT), 12))
     selected_model = str(model or DEFAULT_MODEL)
+    selected_route_host = ""
     selected_timeout = max(10, int(timeout or DEFAULT_TIMEOUT_SECONDS))
     readiness_timeout = max(1, min(int(route_readiness_timeout or 3), 20))
     generate_probe_timeout = max(
@@ -489,6 +589,7 @@ def run_pixelrag_ollama_vlm_replay_worker(
         model_route_ready = bool(route_summary.get("model_route_ready"))
         if candidate_model:
             selected_model = candidate_model
+            selected_route_host = str(route_summary.get("candidate_host") or "").strip()
 
     worker_items: list[dict[str, Any]] = []
     for item in replay_items:
@@ -511,6 +612,7 @@ def run_pixelrag_ollama_vlm_replay_worker(
             root=root,
             output_root=output,
             model=selected_model,
+            route_host=selected_route_host,
             timeout=selected_timeout,
             tile_limit=tiles,
             write_receipt=write_receipt,
@@ -607,6 +709,7 @@ def run_pixelrag_ollama_vlm_replay_worker(
         "tile_limit": tiles,
         "model": selected_model,
         "configured_model": str(model or DEFAULT_MODEL),
+        "route_candidate_host": selected_route_host,
         "timeout_seconds": selected_timeout,
         "execute": bool(execute),
         "write_receipt": bool(write_receipt),