local_server.py

Go to the documentation of this file.

# Copyright (c) 2025-present Aquin Labs Private Limited. All Rights Reserved.
# This file is part of the Aquin Engine. Unauthorized copying, modification,
# distribution, or use of this file, via any medium, is strictly prohibited.
# Proprietary and confidential. See LICENSE for terms.
 
"""
Local HTTP server on 127.0.0.1 — started as a daemon thread during `aquin chat`.
Default port: AQUIN_ENGINE_PORT → ~/.aquin/engine-info.json → 17832 (not 8002).
Handles routes from inspection-backend so the web app can call GPU tools
without a VM.
"""
from __future__ import annotations
 
import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from typing import Any
 
from aquin.engine.engine_info import DEFAULT_ENGINE_PORT, resolve_engine_port
 
PORT = DEFAULT_ENGINE_PORT
_server: HTTPServer | None = None
_thread: threading.Thread | None = None
# True only when this process is the standalone model daemon (run_forever),
# not the short-lived thread server started during `aquin chat`.
_is_daemon: bool = False
# Atomic switch accept (HTTP is usually one-at-a-time; bg tools can interleave).
_switch_accept_lock = threading.Lock()
# Shared label cache (populated by label-features/stream, read by feature-neighbors)
_session_label_cache: dict[tuple[int, str], str] = {}
 
 
def _current_loaded_model_id() -> str | None:
    from aquin.compute.model_runtime import resident_from_cache
 
    resident, _ = resident_from_cache()
    return resident
 
 
def _load_model_in_background(requested_model_id: str) -> None:
    try:
        from aquin.compute.model_runtime import load_weights
        from aquin.engine.engine_info import set_daemon_state
 
        slug = load_weights(requested_model_id)
        try:
            set_daemon_state(status="running", model_id=slug)
        except Exception:
            pass
    except Exception as exc:
        try:
            from aquin.compute.model_runtime import fail_load
 
            fail_load(str(requested_model_id), str(exc) or exc.__class__.__name__)
        except Exception:
            pass
        print(f"[daemon] model load failed: {exc}", flush=True)
 
 
def _read_body(handler: BaseHTTPRequestHandler) -> dict:
    length = int(handler.headers.get("Content-Length", 0))
    raw = handler.rfile.read(length) if length else b"{}"
    try:
        return json.loads(raw)
    except Exception:
        return {}
 
 
def _respond(handler: BaseHTTPRequestHandler, data: Any, status: int = 200) -> None:
    # allow_nan=False path: strip NaN/Inf so Electron/Node never see invalid JSON.
    from aquin.engine.json_safe import dumps_json_safe
 
    body = dumps_json_safe(data).encode()
    handler.send_response(status)
    handler.send_header("Content-Type", "application/json")
    handler.send_header("Content-Length", str(len(body)))
    handler.send_header("Access-Control-Allow-Origin", "*")
    handler.end_headers()
    handler.wfile.write(body)
 
 
def _respond_error(handler: BaseHTTPRequestHandler, msg: str, status: int = 500) -> None:
    _respond(handler, {"error": msg}, status)
 
 
class _Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass  # suppress access log
 
    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "POST, GET, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization, x-tab-id")
        self.end_headers()
 
    def do_GET(self):
        path = self.path.split("?")[0]
        params = {}
        if "?" in self.path:
            import urllib.parse
            params = dict(urllib.parse.parse_qsl(self.path.split("?", 1)[1]))
 
        if path == "/health":
            self._handle_health()
        elif path == "/model/status":
            self._handle_model_status()
        elif path == "/sae-layers":
            self._handle_sae_layers(params)
        elif path == "/model/load":
            self._handle_model_load(params)
        else:
            _respond_error(self, f"Not found: {path}", 404)
 
    def do_POST(self):
        path = self.path.split("?")[0]
 
        handlers = {
            "/feature-logits": self._handle_feature_logits,
            "/feature-neighbors": self._handle_feature_neighbors,
            "/umap": self._handle_umap,
            "/label-features/stream": self._handle_label_features_stream,
            "/features": self._handle_features,
            "/logit-lens": self._handle_logit_lens,
            "/trace": self._handle_trace,
            "/attribution": self._handle_attribution,
            "/interp-score": self._handle_interp_score,
            "/steer": self._handle_steer,
            "/steer-multi": self._handle_steer_multi,
            "/evals/consistency": self._handle_evals_consistency,
            "/evals/suppression": self._handle_evals_suppression,
            "/evals/boundary": self._handle_evals_boundary,
            "/chat/inspect": self._handle_chat_inspect,
            # Persistent model daemon control + generic tool dispatch
            "/dispatch": self._handle_dispatch,
            "/tools/execute": self._handle_dispatch,
            "/model/switch": self._handle_model_switch,
            "/sae/load": self._handle_sae_load,
            "/prompt": self._handle_prompt,
            "/unload": self._handle_unload,
            "/shutdown": self._handle_shutdown,
        }
 
        handler_fn = handlers.get(path)
        if handler_fn:
            handler_fn()
        else:
            _respond_error(self, f"Unknown path: {path}", 404)
 
    # ── GET handlers ──────────────────────────────────────────────────────────
 
    def _handle_health(self):
        from aquin.compute.model_runtime import health_payload
 
        _respond(self, health_payload(daemon=_is_daemon, port=PORT))
 
    def _handle_model_status(self):
        from aquin.compute.model_runtime import health_payload, snapshot
 
        payload = health_payload(daemon=_is_daemon, port=PORT)
        snap = snapshot()
        _respond(
            self,
            {
                "model_id": payload.get("model_id"),
                "daemon": _is_daemon,
                "model_status": payload.get("model_status"),
                "loading_model_id": payload.get("loading_model_id"),
                "last_error": payload.get("last_error"),
                "model_kind": snap.kind,
                "elapsed_s": payload.get("elapsed_s"),
                "vram": payload.get("vram"),
            },
        )
 
    # ── Persistent daemon: generic dispatch + model lifecycle ─────────────────
 
    def _handle_dispatch(self):
        body = _read_body(self)
        name = body.get("name")
        args = body.get("args") or {}
        ctx = body.get("ctx") or {}
        if not name:
            _respond_error(self, "name required", 400)
            return
        try:
            from aquin.engine.tools.registry import dispatch, _load_stubs
            _load_stubs()
            try:
                from aquin.compute.loader_shim import apply as _shim_apply
                _shim_apply()
            except Exception:
                pass
            result = dispatch(name, args, ctx)
            _respond(self, {"ok": True, "result": result})
        except NotImplementedError as e:
            _respond(self, {"ok": False, "error": str(e) or "not implemented"})
        except Exception as e:
            _respond(self, {"ok": False, "error": str(e)})
 
    def _handle_model_switch(self):
        body = _read_body(self)
        model_id = body.get("model_id")
        if not model_id:
            _respond_error(self, "model_id required", 400)
            return
        try:
            from aquin.compute.model_runtime import (
                ModelPhase,
                begin_load,
                finish_load,
                resolve_kind,
                snapshot,
            )
 
            resolved, kind = resolve_kind(model_id)
            with _switch_accept_lock:
                current = snapshot()
                if current.phase == ModelPhase.LOADING:
                    if current.target_id == resolved:
                        _respond(
                            self,
                            {
                                "ok": True,
                                "accepted": True,
                                "status": "loading",
                                "model_id": resolved,
                            },
                        )
                        return
                    _respond(
                        self,
                        {
                            "ok": False,
                            "error": (
                                f"background engine is already loading {current.target_id}. "
                                "Cancel that load first (Remove / Cancel in the Model picker)."
                            ),
                        },
                    )
                    return
 
                loaded = _current_loaded_model_id()
                if loaded == resolved:
                    finish_load(resolved, kind)
                    _respond(
                        self,
                        {
                            "ok": True,
                            "accepted": False,
                            "status": "ready",
                            "model_id": resolved,
                        },
                    )
                    return
 
                # Mark LOADING before spawn so concurrent switch/tools see it.
                begin_load(model_id)
 
            threading.Thread(
                target=_load_model_in_background,
                args=(model_id,),
                daemon=True,
            ).start()
            _respond(
                self,
                {"ok": True, "accepted": True, "status": "loading", "model_id": resolved},
            )
        except Exception as e:
            _respond(self, {"ok": False, "error": str(e)})
 
    def _handle_sae_load(self):
        body = _read_body(self)
        model_id = body.get("model_id")
        layer = body.get("layer")
        if not model_id or layer is None:
            _respond_error(self, "model_id and layer required", 400)
            return
        try:
            from aquin.compute.model_loader import (
                resolve_model_id, get_loaded_model, load_model, load_sae, evict_sae_cache,
            )
            mid = resolve_model_id(model_id)
            evict_sae_cache(mid, int(layer))  # force a fresh read after a (re)bind
            model = get_loaded_model() or load_model(mid)
            load_sae(model, int(layer), mid)
            _respond(self, {"ok": True, "model_id": mid, "layer": int(layer)})
        except Exception as e:
            _respond(self, {"ok": False, "error": str(e)})
 
    def _handle_prompt(self):
        """Quick try-out completion against the resident model."""
        body = _read_body(self)
        prompt = (body.get("prompt") or "").strip()
        if not prompt:
            _respond_error(self, "prompt required", 400)
            return
        try:
            from aquin.compute.causal_trace import run_chat
            from aquin.compute.model_loader import get_active_model_id
 
            model_id = (body.get("model_id") or "").strip() or _current_loaded_model_id() or get_active_model_id()
            if not model_id:
                _respond(self, {"ok": False, "error": "no model loaded — run: aquin load model <id>"})
                return
            max_new = int(body.get("max_new_tokens") or body.get("max_tokens") or 200)
            temperature = float(body.get("temperature", 0.7))
            response = run_chat(
                prompt,
                model_id=model_id,
                max_new_tokens=max_new,
                temperature=temperature,
            )
            _respond(
                self,
                {
                    "ok": True,
                    "model_id": model_id,
                    "prompt": prompt,
                    "response": response,
                },
            )
        except Exception as e:
            _respond(self, {"ok": False, "error": str(e)})
 
    def _handle_unload(self):
        try:
            from aquin.compute.model_runtime import unload_weights
            from aquin.engine.engine_info import set_daemon_state
 
            unload_weights()
            try:
                set_daemon_state(status="running", model_id=None)
            except Exception:
                pass
            _respond(self, {"ok": True})
        except Exception as e:
            _respond(self, {"ok": False, "error": str(e)})
 
    def _handle_shutdown(self):
        _respond(self, {"ok": True})
        threading.Thread(target=stop, daemon=True).start()
 
    def _handle_sae_layers(self, params: dict):
        model_id = params.get("model_id", "llama-3.2-1b")
        try:
            from aquin.compute.model_loader import get_config, resolve_model_id
            model_id = resolve_model_id(model_id)
            cfg = get_config(model_id)
            layer = cfg.get("sae_layer", 8)
            _respond(self, {"model_id": model_id, "layers": [layer], "default": layer})
        except Exception as e:
            _respond_error(self, str(e))
 
    def _handle_model_load(self, params: dict):
        model_id = params.get("model_id", "llama-3.2-1b")
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Cache-Control", "no-cache")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
 
        def _sse(data: dict) -> None:
            line = f"data: {json.dumps(data)}\n\n"
            self.wfile.write(line.encode())
            self.wfile.flush()
 
        try:
            from aquin.compute.model_runtime import ModelPhase, load_weights, snapshot
            from aquin.compute.model_loader import resolve_model_id
 
            model_id = resolve_model_id(model_id)
            snap = snapshot()
            if snap.phase == ModelPhase.LOADING and snap.target_id != model_id:
                _sse(
                    {
                        "stage": "error",
                        "message": (
                            f"already loading {snap.target_id}; cancel first"
                        ),
                    }
                )
                return
            _sse({"stage": "model", "message": f"Loading {model_id}..."})
            slug = load_weights(model_id)
            _sse({"stage": "done", "model_id": slug})
        except Exception as e:
            _sse({"stage": "error", "message": str(e)})
 
    # ── POST handlers ─────────────────────────────────────────────────────────
 
    def _handle_feature_logits(self):
        body = _read_body(self)
        feature_idx = body.get("feature_idx")
        model_id = body.get("model_id", "llama-3.2-1b")
        layer = body.get("layer")
        top_k = int(body.get("top_k", 10))
 
        if feature_idx is None:
            _respond_error(self, "feature_idx required", 400)
            return
 
        try:
            from aquin.compute.model_loader import get_loaded_model, load_model, resolve_model_id
            from aquin.compute.feature_analysis import get_feature_logits
 
            model_id = resolve_model_id(model_id)
            model = get_loaded_model()
            if model is None:
                model = load_model(model_id)
 
            from aquin.compute.feature_analysis import enrich_feature_tool_result, prompt_for_labeling
            from aquin.compute.openai_client import get_openai_client
 
            result = get_feature_logits(int(feature_idx), model, model_id=model_id, layer=layer, top_k=top_k)
            enrich_feature_tool_result(
                result,
                prompt=prompt_for_labeling(),
                model=model,
                client=get_openai_client(),
                model_id=model_id,
                layer=layer,
            )
            _respond(self, result)
        except Exception as e:
            _respond_error(self, str(e))
 
    def _handle_feature_neighbors(self):
        body = _read_body(self)
        feature_idx = body.get("feature_idx")
        model_id = body.get("model_id", "llama-3.2-1b")
        layer = body.get("layer")
        top_k = int(body.get("top_k", 8))
 
        if feature_idx is None:
            _respond_error(self, "feature_idx required", 400)
            return
 
        try:
            from aquin.compute.model_loader import get_loaded_model, load_model, resolve_model_id
            from aquin.compute.feature_analysis import (
                enrich_feature_tool_result,
                get_feature_neighbors,
                prompt_for_labeling,
            )
            from aquin.compute.openai_client import get_openai_client
 
            model_id = resolve_model_id(model_id)
            model = get_loaded_model()
            if model is None:
                model = load_model(model_id)
            result = get_feature_neighbors(int(feature_idx), model_id=model_id, layer=layer, top_k=top_k)
            enrich_feature_tool_result(
                result,
                prompt=prompt_for_labeling(),
                model=model,
                client=get_openai_client(),
                model_id=model_id,
                layer=layer,
                label_neighbors=True,
            )
            _respond(self, result)
        except Exception as e:
            _respond_error(self, str(e))
 
    def _handle_umap(self):
        body = _read_body(self)
        model_id = body.get("model_id", "llama-3.2-1b")
        layer = body.get("layer")
 
        try:
            from aquin.compute.bridge import _ensure_umap_loaded
 
            result = _ensure_umap_loaded({"model_id": model_id, "layer": layer}, {})
            if result.get("error"):
                _respond_error(self, str(result["error"]), 500)
                return
            _respond(self, {"points": result.get("points", []), "cached": False})
        except Exception as e:
            _respond_error(self, str(e))
 
    def _handle_label_features_stream(self):
        body = _read_body(self)
        prompt = body.get("prompt", "")
        response = body.get("response", "")
        model_id = body.get("model_id", "llama-3.2-1b")
        layer = body.get("layer")
 
        try:
            from aquin.compute.model_loader import get_loaded_model, load_model, resolve_model_id
            from aquin.compute.feature_analysis import label_inspection_features, run_feature_analysis_unlabeled
            from aquin.compute.openai_client import get_openai_client
 
            model_id = resolve_model_id(model_id)
            model = get_loaded_model()
            if model is None:
                model = load_model(model_id)
 
            feat = run_feature_analysis_unlabeled(prompt, response, model, model_id=model_id, layer=layer)
            label_inspection_features(
                feat,
                prompt=prompt,
                model=model,
                client=get_openai_client(),
                model_id=model_id,
                layer=layer,
            )
 
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Cache-Control", "no-cache")
            self.send_header("Access-Control-Allow-Origin", "*")
            self.end_headers()
 
            def _sse(data: dict) -> None:
                line = f"data: {json.dumps(data)}\n\n"
                self.wfile.write(line.encode())
                self.wfile.flush()
 
            _sse({"type": "tokens", **feat})
 
            seen: set[int] = set()
            for f in feat.get("top_response_features", []):
                fi = f["feature_idx"]
                if fi not in seen:
                    seen.add(fi)
                    _sse({"type": "label", "feature_idx": fi, "label": f.get("label", f"feature_{fi}")})
            for attr in feat.get("attribution", []):
                for f in attr.get("driven_by_features", []):
                    fi = f["feature_idx"]
                    if fi not in seen:
                        seen.add(fi)
                        _sse({"type": "label", "feature_idx": fi, "label": f.get("label", f"feature_{fi}")})
 
            _sse({"type": "done"})
 
        except Exception as e:
            self.send_response(500)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps({"error": str(e)}).encode())
 
    def _handle_features(self):
        body = _read_body(self)
        prompt = body.get("prompt", "")
        response = body.get("response", "")
        model_id = body.get("model_id", "llama-3.2-1b")
        layer = body.get("layer")
 
        try:
            from aquin.compute.model_loader import get_loaded_model, load_model, resolve_model_id
            from aquin.compute.feature_analysis import run_feature_analysis_unlabeled
 
            model_id = resolve_model_id(model_id)
            model = get_loaded_model()
            if model is None:
                model = load_model(model_id)
 
            result = run_feature_analysis_unlabeled(prompt, response, model, model_id=model_id, layer=layer)
            _respond(self, result)
        except Exception as e:
            _respond_error(self, str(e))
 
    def _handle_logit_lens(self):
        body = _read_body(self)
        prompt = body.get("prompt", "")
        top_k = int(body.get("top_k", 5))
        model_id = body.get("model_id", "llama-3.2-1b")
 
        try:
            from aquin.compute.model_loader import resolve_model_id
            from aquin.compute.causal_trace import run_logit_lens
 
            model_id = resolve_model_id(model_id)
            results = run_logit_lens(prompt, model_id=model_id, top_k=top_k)
            _respond(self, {"results": results})
        except Exception as e:
            _respond_error(self, str(e))
 
    def _handle_trace(self):
        body = _read_body(self)
        prompt = body.get("prompt", "")
        target = body.get("target", "")
        model_id = body.get("model_id", "llama-3.2-1b")
 
        try:
            from aquin.compute.model_loader import resolve_model_id
            from aquin.compute.causal_trace import run_trace
 
            model_id = resolve_model_id(model_id)
            results = run_trace(prompt, target, model_id=model_id)
            _respond(self, {"results": results})
        except Exception as e:
            _respond_error(self, str(e))
 
    def _handle_attribution(self):
        body = _read_body(self)
        prompt = body.get("prompt", "")
        response = body.get("response", "")
        model_id = body.get("model_id", "llama-3.2-1b")
 
        prompt_tokens = prompt.split(" ")
        response_tokens = response.split(" ")
        sig_prompt_tis = list(range(len(prompt_tokens)))
        sig_response_tis = list(range(len(response_tokens)))
        noise_scale = float(body.get("noise_scale", 3.0))
        n_noise_runs = int(body.get("n_noise_runs", 5))
 
        if "prompt_tokens" in body:
            prompt_tokens = body["prompt_tokens"]
            response_tokens = body["response_tokens"]
            sig_prompt_tis = body.get("sig_prompt_tis", sig_prompt_tis)
            sig_response_tis = body.get("sig_response_tis", sig_response_tis)
 
        try:
            from aquin.compute.model_loader import resolve_model_id
            from aquin.compute.causal_trace import run_prompt_attribution
 
            model_id = resolve_model_id(model_id)
            result = run_prompt_attribution(
                prompt, response,
                prompt_tokens, response_tokens,
                sig_prompt_tis, sig_response_tis,
                model_id,
                noise_scale, n_noise_runs,
            )
            _respond(self, result)
        except Exception as e:
            _respond_error(self, str(e))
 
    def _handle_interp_score(self):
        body = _read_body(self)
        feature_idx = body.get("feature_idx")
        prompt = body.get("prompt", "")
        n_samples = int(body.get("n_samples", 10))
        layer = body.get("layer")
        model_id = body.get("model_id", "llama-3.2-1b")
 
        if feature_idx is None:
            _respond_error(self, "feature_idx required", 400)
            return
 
        # interp_score requires OpenAI — not available on user machine
        _respond_error(self, "interp-score requires OpenAI API key (server-side only)", 503)
 
    def _handle_steer(self):
        body = _read_body(self)
        prompt = body.get("prompt", "")
        feature_idx = body.get("feature_idx")
        strength = float(body.get("strength", 20.0))
        model_id = body.get("model_id", "llama-3.2-1b")
        max_new_tokens = int(body.get("max_new_tokens", 200))
        temperature = float(body.get("temperature", 0.7))
 
        if feature_idx is None:
            _respond_error(self, "feature_idx required", 400)
            return
 
        try:
            from aquin.compute.model_loader import get_loaded_model, load_model, resolve_model_id
            from aquin.compute.steer import run_steer_stream
 
            model_id = resolve_model_id(model_id)
            model = get_loaded_model()
            if model is None:
                model = load_model(model_id)
 
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Cache-Control", "no-cache")
            self.send_header("Access-Control-Allow-Origin", "*")
            self.end_headers()
 
            for event in run_steer_stream(
                prompt=prompt,
                feature_idx=int(feature_idx),
                strength=strength,
                model_id=model_id,
                max_new_tokens=max_new_tokens,
                temperature=temperature,
            ):
                line = f"data: {json.dumps(event)}\n\n"
                self.wfile.write(line.encode())
                self.wfile.flush()
 
        except Exception as e:
            try:
                self.wfile.write(f"data: {json.dumps({'error': str(e)})}\n\n".encode())
                self.wfile.flush()
            except Exception:
                pass
 
    def _handle_steer_multi(self):
        body = _read_body(self)
        _respond_error(self, "steer-multi not yet implemented in local engine", 501)
 
    def _handle_evals_consistency(self):
        body = _read_body(self)
        query = body.get("query", "")
        templates = body.get("templates")
        model_id = body.get("model_id", "llama-3.2-1b")
 
        if not query:
            _respond_error(self, "query required", 400)
            return
 
        try:
            from aquin.compute.model_loader import get_loaded_model, load_model, resolve_model_id
            from aquin.compute.evals import consistency_eval
 
            model_id = resolve_model_id(model_id)
            model = get_loaded_model()
            if model is None:
                model = load_model(model_id)
 
            result = consistency_eval(query, model, templates=templates or None)
            _respond(self, result)
        except Exception as e:
            _respond_error(self, str(e))
 
    def _handle_evals_suppression(self):
        body = _read_body(self)
        topics = body.get("topics")
        model_id = body.get("model_id", "llama-3.2-1b")
 
        try:
            from aquin.compute.model_loader import get_loaded_model, load_model, resolve_model_id
            from aquin.compute.evals import suppression_eval
 
            model_id = resolve_model_id(model_id)
            model = get_loaded_model()
            if model is None:
                model = load_model(model_id)
 
            result = suppression_eval(model, topics=topics or None)
            _respond(self, result)
        except Exception as e:
            _respond_error(self, str(e))
 
    def _handle_evals_boundary(self):
        body = _read_body(self)
        prompts = body.get("prompts", [])
        model_id = body.get("model_id", "llama-3.2-1b")
 
        if not prompts:
            _respond_error(self, "prompts list cannot be empty", 400)
            return
        if len(prompts) > 20:
            _respond_error(self, "max 20 prompts per boundary eval run", 400)
            return
 
        try:
            from aquin.compute.model_loader import get_loaded_model, load_model, resolve_model_id
            from aquin.compute.evals import boundary_eval
 
            model_id = resolve_model_id(model_id)
            model = get_loaded_model()
            if model is None:
                model = load_model(model_id)
 
            result = boundary_eval(prompts, model)
            _respond(self, result)
        except Exception as e:
            _respond_error(self, str(e))
 
    def _handle_chat_inspect(self):
        body = _read_body(self)
        messages = body.get("messages", [])
        model_id = body.get("model_id", "llama-3.2-1b")
        max_new_tokens = int(body.get("max_new_tokens", 200))
        temperature = float(body.get("temperature", 0.7))
 
        user_content = ""
        for m in reversed(messages):
            if m.get("role") == "user":
                user_content = m.get("content", "")
                break
 
        try:
            from aquin.compute.model_loader import resolve_model_id
            from aquin.compute.causal_trace import run_chat
 
            model_id = resolve_model_id(model_id)
            response = run_chat(user_content, model_id=model_id, max_new_tokens=max_new_tokens, temperature=temperature)
            _respond(self, {"response": response})
        except Exception as e:
            _respond_error(self, str(e))
 
 
def start() -> None:
    global _server, _thread, PORT
    if _server is not None:
        return
 
    PORT = resolve_engine_port()
    try:
        _server = HTTPServer(("127.0.0.1", PORT), _Handler)
    except OSError:
        # Port already in use — another aquin chat instance running
        return
 
    _thread = threading.Thread(target=_server.serve_forever, daemon=True)
    _thread.start()
    print("[syncing with web]", flush=True)
 
 
def stop() -> None:
    global _server
    if _server:
        _server.shutdown()
        _server = None
 
 
def run_forever(port: int = PORT) -> None:
    """Run the server in the foreground (used by the standalone model daemon)."""
    global _server, _is_daemon
    _is_daemon = True
    try:
        _server = HTTPServer(("127.0.0.1", port), _Handler)
    except OSError as exc:
        print(f"[daemon] cannot bind 127.0.0.1:{port}: {exc}", flush=True)
        return
    print(f"[daemon] listening on 127.0.0.1:{port}", flush=True)
    try:
        _server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        _server = None