tenseleyflow/documentlanguagemodel / 82b797f

+`llama-server`.

+| `--target NAME` | `ollama` | Export destination. Sprint 41 currently supports `ollama` and `llama-server`. The `llama-server` path writes launch artifacts against the existing GGUF export and uses the shared OpenAI-compatible HTTP smoke harness when `--no-smoke` is not passed. |

+        llama_server_smoke = None if no_smoke else resolved_target.smoke_test(llama_server_result)

+        if llama_server_smoke is not None and not llama_server_smoke.ok:

+            console.print(

+                f"[red]smoke:[/red] {llama_server_smoke.detail}\n"

+                "  re-run with `--no-smoke` to skip the smoke test."

+            )

+            raise typer.Exit(code=1)

+        if llama_server_smoke is not None and llama_server_smoke.detail:

+            console.print(f"smoke:   {llama_server_smoke.detail}")

+class TargetSmokeError(ExportError):

+    """A runtime-target smoke check failed to start or answer correctly."""

+

+

+"""Shared HTTP smoke helpers for OpenAI-compatible local runtimes."""

+

+from __future__ import annotations

+

+import json

+import socket

+import subprocess  # nosec B404

+import tempfile

+import time

+import urllib.error

+import urllib.request

+from collections.abc import Sequence

+from typing import TextIO

+

+from dlm.export.errors import TargetSmokeError

+

+_DEFAULT_HOST = "127.0.0.1"

+_DEFAULT_STARTUP_TIMEOUT_SECONDS = 30.0

+_DEFAULT_REQUEST_TIMEOUT_SECONDS = 5.0

+_DEFAULT_POLL_INTERVAL_SECONDS = 0.1

+_DEFAULT_PROMPT = "Hello."

+

+

+def reserve_local_port(host: str = _DEFAULT_HOST) -> int:

+    """Ask the OS for a free loopback TCP port."""

+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:

+        sock.bind((host, 0))

+        return int(sock.getsockname()[1])

+

+

+def smoke_openai_compat_server(

+    command: Sequence[str],

+    *,

+    host: str = _DEFAULT_HOST,

+    port: int | None = None,

+    startup_timeout: float = _DEFAULT_STARTUP_TIMEOUT_SECONDS,

+    request_timeout: float = _DEFAULT_REQUEST_TIMEOUT_SECONDS,

+    poll_interval: float = _DEFAULT_POLL_INTERVAL_SECONDS,

+    prompt: str = _DEFAULT_PROMPT,

+) -> str:

+    """Start a local OpenAI-compatible server, wait for readiness, then chat."""

+

+    real_port = port if port is not None else reserve_local_port(host)

+    argv = _replace_or_append_flag(list(command), "--host", host)

+    argv = _replace_or_append_flag(argv, "--port", str(real_port))

+

+    with tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as log:

+        proc = subprocess.Popen(  # nosec B603

+            argv,

+            stdout=log,

+            stderr=subprocess.STDOUT,

+            text=True,

+        )

+        try:

+            model_id = _wait_for_models(

+                proc,

+                log,

+                host=host,

+                port=real_port,

+                startup_timeout=startup_timeout,

+                request_timeout=request_timeout,

+                poll_interval=poll_interval,

+            )

+            content = _chat_completion(

+                host=host,

+                port=real_port,

+                model_id=model_id,

+                prompt=prompt,

+                request_timeout=request_timeout,

+            )

+            first = _first_non_empty_line(content)

+            if not first:

+                raise TargetSmokeError("openai-compatible smoke returned empty assistant content")

+            return first

+        finally:

+            _stop_process(proc)

+

+

+def _wait_for_models(

+    proc: subprocess.Popen[str],

+    log: TextIO,

+    *,

+    host: str,

+    port: int,

+    startup_timeout: float,

+    request_timeout: float,

+    poll_interval: float,

+) -> str | None:

+    deadline = time.monotonic() + startup_timeout

+    last_error: str | None = None

+    while time.monotonic() < deadline:

+        if proc.poll() is not None:

+            raise TargetSmokeError(

+                f"server exited before readiness (exit {proc.returncode}){_log_tail(log)}"

+            )

+        try:

+            return _fetch_model_id(host=host, port=port, request_timeout=request_timeout)

+        except (

+            OSError,

+            TimeoutError,

+            ValueError,

+            urllib.error.HTTPError,

+            urllib.error.URLError,

+        ) as exc:

+            last_error = f"{type(exc).__name__}: {exc}"

+            time.sleep(poll_interval)

+    suffix = f" last error: {last_error}." if last_error else "."

+    raise TargetSmokeError(

+        f"server did not become ready on http://{host}:{port}/v1/models within "

+        f"{startup_timeout:.1f}s.{suffix}{_log_tail(log)}"

+    )

+

+

+def _fetch_model_id(*, host: str, port: int, request_timeout: float) -> str | None:

+    req = urllib.request.Request(

+        f"http://{host}:{port}/v1/models",

+        headers={"Accept": "application/json"},

+        method="GET",

+    )

+    with urllib.request.urlopen(req, timeout=request_timeout) as resp:  # noqa: S310

+        payload = json.loads(resp.read())

+    data = payload.get("data")

+    if not isinstance(data, list) or not data:

+        return None

+    first = data[0]

+    if not isinstance(first, dict):

+        return None

+    model_id = first.get("id")

+    return model_id if isinstance(model_id, str) and model_id.strip() else None

+

+

+def _chat_completion(

+    *,

+    host: str,

+    port: int,

+    model_id: str | None,

+    prompt: str,

+    request_timeout: float,

+) -> str:

+    payload = {

+        "model": model_id or "dlm-smoke",

+        "messages": [{"role": "user", "content": prompt}],

+    }

+    req = urllib.request.Request(

+        f"http://{host}:{port}/v1/chat/completions",

+        data=json.dumps(payload).encode("utf-8"),

+        headers={

+            "Accept": "application/json",

+            "Content-Type": "application/json",

+        },

+        method="POST",

+    )

+    with urllib.request.urlopen(req, timeout=request_timeout) as resp:  # noqa: S310

+        body = json.loads(resp.read())

+    choices = body.get("choices")

+    if not isinstance(choices, list) or not choices:

+        raise TargetSmokeError("chat completion response missing choices")

+    first_choice = choices[0]

+    if not isinstance(first_choice, dict):

+        raise TargetSmokeError("chat completion response has non-object choices[0]")

+    message = first_choice.get("message")

+    if not isinstance(message, dict):

+        raise TargetSmokeError("chat completion response missing choices[0].message")

+    content = _normalize_message_content(message.get("content"))

+    if content is None:

+        raise TargetSmokeError(

+            "chat completion response missing non-empty choices[0].message.content"

+        )

+    return content

+

+

+def _normalize_message_content(content: object) -> str | None:

+    if isinstance(content, str):

+        stripped = content.strip()

+        return stripped if stripped else None

+    if isinstance(content, list):

+        parts: list[str] = []

+        for item in content:

+            if not isinstance(item, dict):

+                continue

+            text = item.get("text")

+            if isinstance(text, str) and text.strip():

+                parts.append(text.strip())

+        merged = "\n".join(parts).strip()

+        return merged if merged else None

+    return None

+

+

+def _replace_or_append_flag(argv: list[str], flag: str, value: str) -> list[str]:

+    updated = list(argv)

+    try:

+        idx = updated.index(flag)

+    except ValueError:

+        updated.extend([flag, value])

+        return updated

+    if idx + 1 >= len(updated):

+        updated.append(value)

+        return updated

+    updated[idx + 1] = value

+    return updated

+

+

+def _first_non_empty_line(text: str) -> str:

+    for line in text.splitlines():

+        stripped = line.strip()

+        if stripped:

+            return stripped

+    return ""

+

+

+def _stop_process(proc: subprocess.Popen[str]) -> None:

+    if proc.poll() is not None:

+        return

+    proc.terminate()

+    try:

+        proc.wait(timeout=5.0)

+    except subprocess.TimeoutExpired:

+        proc.kill()

+        proc.wait(timeout=5.0)

+

+

+def _log_tail(log: TextIO, *, lines: int = 20) -> str:

+    log.seek(0)

+    text = log.read().strip()

+    if not text:

+        return ""

+    tail = "\n".join(text.splitlines()[-lines:])

+    return f"\n--- server log tail ---\n{tail}"

+from dlm.base_models import BaseModelSpec

+from dlm.export.errors import ExportError, TargetSmokeError

+from dlm.export.smoke import smoke_openai_compat_server

+        training_sequence_len = _optional_int_extra(ctx, "training_sequence_len")

+        spec = _require_spec_extra(ctx, "spec")

+        model_path = _find_artifact(ctx.artifacts, prefix="base.")

+        adapter_gguf_path = _find_optional_artifact(ctx.artifacts, exact_name="adapter.gguf")

+        context_length = resolve_num_ctx(training_sequence_len, spec.context_length)

+        try:

+            first_line = smoke_openai_compat_server(self._runtime_command(prepared))

+        except (OSError, TargetSmokeError, ExportError) as exc:

+            return SmokeResult(attempted=True, ok=False, detail=str(exc))

+        return SmokeResult(attempted=True, ok=True, detail=first_line)

+

+    def _runtime_command(self, prepared: TargetResult) -> list[str]:

+        model_path = _require_prepared_path(prepared, "model_path")

+        adapter_gguf_path = _optional_prepared_path(prepared, "adapter_gguf_path")

+        context_length = _require_prepared_int(prepared, "context_length")

+        vendor_override = _optional_prepared_path(prepared, "vendor_override")

+

+        command = [

+            str(llama_server_bin(vendor_override)),

+            "--model",

+            str(model_path),

+            "--api-key",

+            "disabled",

+            "--ctx-size",

+            str(context_length),

+            "--chat-template-file",

+            str(prepared.config_path),

+            "--host",

+            "127.0.0.1",

+            "--port",

+            "8000",

+        ]

+        if adapter_gguf_path is not None:

+            command.extend(["--lora", str(adapter_gguf_path)])

+        return command

+            "training_sequence_len": training_sequence_len,

+            "spec": spec,

+def _optional_int_extra(ctx: DispatchResult, key: str) -> int | None:

+    if value is None:

+        return None

+        raise ExportError(f"llama-server target extra {key!r} must be an int")

+    return value

+

+

+def _require_spec_extra(ctx: DispatchResult, key: str) -> BaseModelSpec:

+    value = ctx.extras.get(key)

+    if not isinstance(value, BaseModelSpec):

+        raise ExportError(f"llama-server target missing BaseModelSpec extra {key!r}")

+    def test_llama_server_target_reaches_existing_mutex_validation(self, tmp_path: Path) -> None:

+                "--draft",

+                "qwen2.5:0.5b",

+                "--no-draft",

+        text = _joined(result)

+        assert "mutually exclusive" in text

+        assert "--no-smoke" not in text

+"""llama-server smoke wiring."""

+

+from __future__ import annotations

+

+from pathlib import Path

+

+from dlm.export.errors import TargetSmokeError

+from dlm.export.targets.base import TargetResult

+from dlm.export.targets.llama_server import LLAMA_SERVER_TARGET

+

+

+def _vendor_tree(tmp_path: Path) -> Path:

+    vendor = tmp_path / "vendor" / "llama.cpp"

+    (vendor / "build" / "bin").mkdir(parents=True)

+    server = vendor / "build" / "bin" / "llama-server"

+    server.write_text("#!/bin/sh\n", encoding="utf-8")

+    server.chmod(0o755)

+    return vendor

+

+

+def _prepared_target(tmp_path: Path) -> TargetResult:

+    export_dir = tmp_path / "exports" / "Q4_K_M"

+    export_dir.mkdir(parents=True)

+    manifest_path = export_dir / "export_manifest.json"

+    manifest_path.write_text("{}\n", encoding="utf-8")

+    model = export_dir / "base.Q4_K_M.gguf"

+    model.write_bytes(b"base")

+    adapter = export_dir / "adapter.gguf"

+    adapter.write_bytes(b"adapter")

+    template = export_dir / "chat-template.jinja"

+    template.write_text("{{ .Prompt }}\n", encoding="utf-8")

+    return TargetResult(

+        name="llama-server",

+        export_dir=export_dir,

+        manifest_path=manifest_path,

+        artifacts=(model, adapter, template),

+        config_path=template,

+        extras={

+            "model_path": model,

+            "adapter_gguf_path": adapter,

+            "context_length": 4096,

+            "vendor_override": _vendor_tree(tmp_path),

+        },

+    )

+

+

+class TestLlamaServerSmoke:

+    def test_smoke_uses_absolute_runtime_argv(self, tmp_path: Path, monkeypatch: object) -> None:

+        prepared = _prepared_target(tmp_path)

+        seen: list[list[str]] = []

+

+        def _fake_smoke(argv: list[str], **_: object) -> str:

+            seen.append(list(argv))

+            return "server replied"

+

+        monkeypatch.setattr(

+            "dlm.export.targets.llama_server.smoke_openai_compat_server", _fake_smoke

+        )

+

+        result = LLAMA_SERVER_TARGET.smoke_test(prepared)

+

+        assert result.attempted is True

+        assert result.ok is True

+        assert result.detail == "server replied"

+        assert len(seen) == 1

+        argv = seen[0]

+        assert argv[0].endswith("llama-server")

+        assert "$SCRIPT_DIR" not in " ".join(argv)

+        assert str(prepared.extras["model_path"]) in argv

+        assert str(prepared.config_path) in argv

+        assert str(prepared.extras["adapter_gguf_path"]) in argv

+        assert "--host" in argv

+        assert "--port" in argv

+

+    def test_smoke_failure_returns_failed_result(self, tmp_path: Path, monkeypatch: object) -> None:

+        prepared = _prepared_target(tmp_path)

+

+        def _fake_smoke(argv: list[str], **_: object) -> str:

+            _ = argv

+            raise TargetSmokeError("boom")

+

+        monkeypatch.setattr(

+            "dlm.export.targets.llama_server.smoke_openai_compat_server", _fake_smoke

+        )

+

+        result = LLAMA_SERVER_TARGET.smoke_test(prepared)

+

+        assert result.attempted is True

+        assert result.ok is False

+        assert result.detail == "boom"

+"""Shared OpenAI-compatible smoke harness."""

+

+from __future__ import annotations

+

+import socket

+import sys

+from pathlib import Path

+

+import pytest

+

+from dlm.export.errors import TargetSmokeError

+from dlm.export.smoke import smoke_openai_compat_server

+

+

+def _require_loopback_bind() -> None:

+    try:

+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:

+            sock.bind(("127.0.0.1", 0))

+    except PermissionError as exc:

+        pytest.skip(f"loopback bind blocked on this host: {exc}")

+

+

+def _write_server_script(tmp_path: Path, *, mode: str) -> Path:

+    script = tmp_path / f"fake_server_{mode}.py"

+    script.write_text(

+        (

+            "from __future__ import annotations\n"

+            "import argparse\n"

+            "import json\n"

+            "from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer\n"

+            "\n"

+            "parser = argparse.ArgumentParser()\n"

+            "parser.add_argument('--host', required=True)\n"

+            "parser.add_argument('--port', required=True, type=int)\n"

+            "parser.add_argument('--mode', required=True)\n"

+            "args = parser.parse_args()\n"

+            "\n"

+            "if args.mode == 'exit':\n"

+            "    raise SystemExit(3)\n"

+            "\n"

+            "class Handler(BaseHTTPRequestHandler):\n"

+            "    def do_GET(self) -> None:\n"

+            "        if self.path != '/v1/models':\n"

+            "            self.send_response(404)\n"

+            "            self.end_headers()\n"

+            "            return\n"

+            "        body = json.dumps({'data': [{'id': 'fake-model'}]}).encode('utf-8')\n"

+            "        self.send_response(200)\n"

+            "        self.send_header('Content-Type', 'application/json')\n"

+            "        self.send_header('Content-Length', str(len(body)))\n"

+            "        self.end_headers()\n"

+            "        self.wfile.write(body)\n"

+            "\n"

+            "    def do_POST(self) -> None:\n"

+            "        if self.path != '/v1/chat/completions':\n"

+            "            self.send_response(404)\n"

+            "            self.end_headers()\n"

+            "            return\n"

+            "        _ = self.rfile.read(int(self.headers.get('Content-Length', '0')))\n"

+            "        if args.mode == 'empty':\n"

+            "            payload = {'choices': [{'message': {'content': ''}}]}\n"

+            "        else:\n"

+            "            payload = {'choices': [{'message': {'content': 'hello from fake server'}}]}\n"

+            "        body = json.dumps(payload).encode('utf-8')\n"

+            "        self.send_response(200)\n"

+            "        self.send_header('Content-Type', 'application/json')\n"

+            "        self.send_header('Content-Length', str(len(body)))\n"

+            "        self.end_headers()\n"

+            "        self.wfile.write(body)\n"

+            "\n"

+            "    def log_message(self, format: str, *args: object) -> None:\n"

+            "        return\n"

+            "\n"

+            "server = ThreadingHTTPServer((args.host, args.port), Handler)\n"

+            "server.serve_forever()\n"

+        ),

+        encoding="utf-8",

+    )

+    return script

+

+

+class TestSmokeOpenAiCompatServer:

+    def test_returns_first_response_line(self, tmp_path: Path) -> None:

+        _require_loopback_bind()

+        script = _write_server_script(tmp_path, mode="ok")

+

+        first_line = smoke_openai_compat_server(

+            [sys.executable, str(script), "--mode", "ok", "--host", "127.0.0.1", "--port", "8000"]

+        )

+

+        assert first_line == "hello from fake server"

+

+    def test_empty_content_raises(self, tmp_path: Path) -> None:

+        _require_loopback_bind()

+        script = _write_server_script(tmp_path, mode="empty")

+

+        with pytest.raises(TargetSmokeError, match="non-empty"):

+            smoke_openai_compat_server(

+                [

+                    sys.executable,

+                    str(script),

+                    "--mode",

+                    "empty",

+                    "--host",

+                    "127.0.0.1",

+                    "--port",

+                    "8000",

+                ]

+            )

+

+    def test_early_exit_raises_with_readiness_message(self, tmp_path: Path) -> None:

+        _require_loopback_bind()

+        script = _write_server_script(tmp_path, mode="exit")

+

+        with pytest.raises(TargetSmokeError, match="exited before readiness"):

+            smoke_openai_compat_server(

+                [

+                    sys.executable,

+                    str(script),

+                    "--mode",

+                    "exit",

+                    "--host",

+                    "127.0.0.1",

+                    "--port",

+                    "8000",

+                ],

+                startup_timeout=1.0,

+            )