Code reference

This page is generated from the source docstrings at build time (mkdocstrings + griffe, static analysis — no imports are executed). They can't drift from the code, because they are the code. This is a curated tour of the modules worth knowing; the full source is the ultimate reference.

Note

If a symbol below has thin documentation, the fix is to improve its docstring in the source — not to edit this page. The page regenerates on the next build.

---

Domain — errors & pure types

The dependency-free core. Every exception a service raises is one of these; the boundary handler in adapta/api/app.py is the only place they're serialized.

Typed error taxonomy for Adapta.

Rules enforced at the boundary (adapta/api/app.py): - Code raises DomainError subclasses — never HTTPException, never str(e). - internal_detail is logged server-side and never serialized to clients. - Every error response carries a correlation ID stamped by middleware. - make check-leaks CI gate prevents reintroducing raw str(e) to clients.

DomainError dataclass

Bases: Exception

Base class for all application errors.

Fields

message: Safe, human-readable text sent to the client. code: Stable machine-readable error code (snake_case). status: HTTP status code for the boundary handler. internal_detail: Diagnostic text logged server-side only — never sent to clients.

The boundary handler in adapta/api/app.py is the only place that translates these into HTTP responses. All service code raises subclasses; nothing outside app.py should catch and re-raise as HTTPException.

Source code in adapta/domain/errors.py

@dataclass
class DomainError(Exception):
    """Base class for all application errors.

    Fields:
        message: Safe, human-readable text sent to the client.
        code: Stable machine-readable error code (snake_case).
        status: HTTP status code for the boundary handler.
        internal_detail: Diagnostic text logged server-side only — never sent to clients.

    The boundary handler in ``adapta/api/app.py`` is the only place that
    translates these into HTTP responses.  All service code raises subclasses;
    nothing outside ``app.py`` should catch and re-raise as HTTPException.
    """

    message: str
    code: str = "internal_error"
    status: int = 500
    internal_detail: Optional[str] = None

    def __str__(self) -> str:
        return self.message

InvalidRequest dataclass

Bases: DomainError

The client sent a semantically invalid value (HTTP 400).

Use this for domain-level validation failures — e.g. a password that is too short, an unsupported file type, or a conflicting state transition. Structural schema failures (missing fields, wrong types) are handled by FastAPI's RequestValidationError and produce 422 automatically.

Source code in adapta/domain/errors.py

@dataclass
class InvalidRequest(DomainError):
    """The client sent a semantically invalid value (HTTP 400).

    Use this for domain-level validation failures — e.g. a password that is
    too short, an unsupported file type, or a conflicting state transition.
    Structural schema failures (missing fields, wrong types) are handled by
    FastAPI's RequestValidationError and produce 422 automatically.
    """

    code: str = field(default="invalid_request", init=False)
    status: int = field(default=400, init=False)

Unauthorized dataclass

Bases: DomainError

The request is missing valid credentials (HTTP 401).

Tells clients to authenticate; do not use when the caller is authenticated but lacks rights — use Forbidden instead.

Source code in adapta/domain/errors.py

@dataclass
class Unauthorized(DomainError):
    """The request is missing valid credentials (HTTP 401).

    Tells clients to authenticate; do not use when the caller is authenticated
    but lacks rights — use Forbidden instead.
    """

    code: str = field(default="unauthorized", init=False)
    status: int = field(default=401, init=False)

Forbidden dataclass

Bases: DomainError

Authenticated caller lacks permission for this resource (HTTP 403).

Distinct from Unauthorized: the token is valid, but the role or team membership check failed. Returning 401 here would wrongly prompt clients to re-authenticate.

Source code in adapta/domain/errors.py

@dataclass
class Forbidden(DomainError):
    """Authenticated caller lacks permission for this resource (HTTP 403).

    Distinct from Unauthorized: the token is valid, but the role or team
    membership check failed.  Returning 401 here would wrongly prompt clients
    to re-authenticate.
    """

    code: str = field(default="forbidden", init=False)
    status: int = field(default=403, init=False)

NotFound dataclass

Bases: DomainError

A requested resource does not exist or is not accessible (HTTP 404).

Source code in adapta/domain/errors.py

@dataclass
class NotFound(DomainError):
    """A requested resource does not exist or is not accessible (HTTP 404)."""

    code: str = field(default="not_found", init=False)
    status: int = field(default=404, init=False)

Conflict dataclass

Bases: DomainError

The request conflicts with existing state (HTTP 409).

Examples: registering a duplicate email, creating a second endpoint for a project that already has one.

Source code in adapta/domain/errors.py

@dataclass
class Conflict(DomainError):
    """The request conflicts with existing state (HTTP 409).

    Examples: registering a duplicate email, creating a second endpoint for a
    project that already has one.
    """

    code: str = field(default="conflict", init=False)
    status: int = field(default=409, init=False)

ModelNotFound dataclass

Bases: NotFound

A referenced base model is not in the catalog or its GGUF is absent (HTTP 404).

Distinct from plain NotFound so callers can differentiate a missing model from a missing project or file.

Source code in adapta/domain/errors.py

@dataclass
class ModelNotFound(NotFound):
    """A referenced base model is not in the catalog or its GGUF is absent (HTTP 404).

    Distinct from plain NotFound so callers can differentiate a missing model
    from a missing project or file.
    """

    code: str = field(default="model_not_found", init=False)

ProjectNotFound dataclass

Bases: NotFound

The referenced project does not exist or the caller's team does not own it.

Source code in adapta/domain/errors.py

@dataclass
class ProjectNotFound(NotFound):
    """The referenced project does not exist or the caller's team does not own it."""

    code: str = field(default="project_not_found", init=False)

TrainingFailed dataclass

Bases: DomainError

A training job failed for a non-eval reason (HTTP 500).

Examples: OOM during QLoRA, broken dataset file on disk, adapter conversion subprocess crash. Distinct from EvalGateFailed (the job ran to completion but the adapter did not clear the threshold).

Source code in adapta/domain/errors.py

@dataclass
class TrainingFailed(DomainError):
    """A training job failed for a non-eval reason (HTTP 500).

    Examples: OOM during QLoRA, broken dataset file on disk, adapter
    conversion subprocess crash.  Distinct from EvalGateFailed (the job ran
    to completion but the adapter did not clear the threshold).
    """

    code: str = field(default="training_failed", init=False)
    status: int = field(default=500, init=False)

InferenceFailed dataclass

Bases: DomainError

The llama-cpp or vLLM inference call raised an unexpected exception (HTTP 500).

Source code in adapta/domain/errors.py

@dataclass
class InferenceFailed(DomainError):
    """The llama-cpp or vLLM inference call raised an unexpected exception (HTTP 500)."""

    code: str = field(default="inference_failed", init=False)
    status: int = field(default=500, init=False)

AdapterConversionFailed dataclass

Bases: DomainError

PEFT→GGUF LoRA conversion failed; the adapter cannot be served (HTTP 500).

Source code in adapta/domain/errors.py

@dataclass
class AdapterConversionFailed(DomainError):
    """PEFT→GGUF LoRA conversion failed; the adapter cannot be served (HTTP 500)."""

    code: str = field(default="adapter_conversion_failed", init=False)
    status: int = field(default=500, init=False)

EmbeddingFailed dataclass

Bases: DomainError

sentence-transformers embedding or cross-encoder reranking call failed (HTTP 500).

Raised from both EmbeddingService and RerankerService so they share the same HTTP surface.

Source code in adapta/domain/errors.py

@dataclass
class EmbeddingFailed(DomainError):
    """sentence-transformers embedding or cross-encoder reranking call failed (HTTP 500).

    Raised from both EmbeddingService and RerankerService so they share the
    same HTTP surface.
    """

    code: str = field(default="embedding_failed", init=False)
    status: int = field(default=500, init=False)

EvalGateFailed dataclass

Bases: DomainError

A trained adapter did not clear the eval threshold (HTTP 422).

HTTP 422 signals a business-rule rejection, not a server error: the job ran successfully, but the resulting adapter is not good enough to serve. The caller may inspect eval_score / eval_metrics on the job, improve the dataset, and re-train.

Source code in adapta/domain/errors.py

@dataclass
class EvalGateFailed(DomainError):
    """A trained adapter did not clear the eval threshold (HTTP 422).

    HTTP 422 signals a business-rule rejection, not a server error: the job
    ran successfully, but the resulting adapter is not good enough to serve.
    The caller may inspect eval_score / eval_metrics on the job, improve the
    dataset, and re-train.
    """

    code: str = field(default="eval_gate_failed", init=False)
    status: int = field(default=422, init=False)

RateLimited dataclass

Bases: DomainError

Too many requests from this client (HTTP 429).

Source code in adapta/domain/errors.py

@dataclass
class RateLimited(DomainError):
    """Too many requests from this client (HTTP 429)."""

    code: str = field(default="rate_limited", init=False)
    status: int = field(default=429, init=False)

Timeout dataclass

Bases: DomainError

An upstream operation (Chroma retrieval, inference) exceeded its deadline (HTTP 504).

504 Gateway Timeout is used rather than 408 Request Timeout because the deadline belongs to the server's call to a downstream component, not to the client's request delivery.

Source code in adapta/domain/errors.py

@dataclass
class Timeout(DomainError):
    """An upstream operation (Chroma retrieval, inference) exceeded its deadline (HTTP 504).

    504 Gateway Timeout is used rather than 408 Request Timeout because the
    deadline belongs to the server's call to a downstream component, not to
    the client's request delivery.
    """

    code: str = field(default="timeout", init=False)
    status: int = field(default=504, init=False)

InternalError dataclass

Bases: DomainError

A named raising point for unexpected server-side conditions (HTTP 500).

Prefer a more specific subclass when the failure mode is known. This class exists so services can write raise InternalError(...) to make intent explicit, rather than instantiating the base DomainError directly.

Source code in adapta/domain/errors.py

@dataclass
class InternalError(DomainError):
    """A named raising point for unexpected server-side conditions (HTTP 500).

    Prefer a more specific subclass when the failure mode is known.  This
    class exists so services can write ``raise InternalError(...)`` to make
    intent explicit, rather than instantiating the base ``DomainError`` directly.
    """

    code: str = field(default="internal_error", init=False)
    status: int = field(default=500, init=False)

---

Configuration

All settings (DB/Redis URLs, JWT secret, directories, thresholds, the eval-gate knobs) flow through one pydantic-settings object.

Application configuration via pydantic-settings. All env vars are prefixed ADAPTA_ (e.g. ADAPTA_DATABASE_URL).

Settings

Bases: BaseSettings

Source code in adapta/config.py

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="ADAPTA_", env_file=".env")

    # Server
    host: str = "0.0.0.0"
    port: int = 8000
    workers: int = 1
    reload: bool = False

    # Paths — set ADAPTA_DATA_DIR to override the root; subdirs default to
    # data_dir/{models,uploads,adapters,datasets} but can be overridden individually.
    data_dir: Path = Path("/app/data")
    models_dir: Path = Path("/app/data/models")
    uploads_dir: Path = Path("/app/data/uploads")
    adapters_dir: Path = Path("/app/data/adapters")
    datasets_dir: Path = Path("/app/data/datasets")

    # Database (Postgres via asyncpg)
    database_url: str = "postgresql+asyncpg://adapta:adapta@postgres:5432/adapta"

    # Redis (job queue + cache)
    redis_url: str = "redis://redis:6379/0"

    # Auth (JWT)
    secret_key: str = "CHANGE_ME_IN_PRODUCTION_use_openssl_rand_hex_32"
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 60 * 8  # 8 hours

    # Development seed (adapta/db/seed.py, run by the entrypoint after migrations).
    # Only acts on an EMPTY database — a real register/bootstrap is never clobbered.
    # The compose file turns it on; disable with ADAPTA_SEED_DEFAULT_ADMIN=0 to
    # exercise the register bootstrap flow instead.
    seed_default_admin: bool = False
    default_admin_org: str = "Local Dev"
    default_admin_email: str = "admin@example.com"
    default_admin_password: str = "admin12345"

    # Inference
    default_model: str = "qwen2.5-3b-instruct"
    max_context_length: int = 4096
    n_threads: int = 8
    n_gpu_layers: int = 0  # 0 = CPU only; >0 = GPU layers
    temperature: float = 0.7
    top_p: float = 0.9
    top_k: int = 40
    max_tokens: int = 512
    # Max characters across all user messages in a single chat request (7.4).
    # Guards against unreasonably large inputs that could exhaust memory during
    # tokenization or consume excessive model context.
    max_input_chars: int = 100_000
    use_mmap: bool = True
    use_mlock: bool = False

    # Inference concurrency & safety (A4.1). llama-cpp's Llama object is NOT safe
    # for concurrent calls on one instance — two requests sharing it race the KV
    # cache (garbage output or a segfault that kills the app). Serving therefore
    # serializes per model (a lock in model_manager) and runs the blocking call on
    # a bounded pool; a per-request wall-clock cap stops a runaway generation from
    # pinning a worker forever.
    inference_max_workers: int = 2
    inference_timeout_seconds: int = 300
    # Max distinct models (base, or base+LoRA) kept loaded at once (A4.8). Since the
    # cache key includes the adapter, N fine-tune endpoints would otherwise pin N full
    # models in RAM and OOM the app container. The LRU is evicted past this bound; an
    # evicted endpoint transparently reloads on its next call. Raise it only with RAM
    # to spare (a 3B Q4 model is ~2 GB resident).
    max_loaded_models: int = 2

    # RAG / embeddings
    embedding_model: str = "all-MiniLM-L6-v2"
    chroma_host: str = "chroma"
    chroma_port: int = 8000
    rag_top_k: int = 5
    # Hybrid retrieval (D5): fetch top_k * N vector candidates before BM25 fusion
    # so the keyword signal has a larger pool to reorder. 4× keeps latency low on CPU.
    rag_hybrid_fetch_multiplier: int = 4
    # Cross-encoder reranker model (D5). Set to "" to disable; rag_timeout_seconds
    # guards the whole retrieval path including reranking.
    rag_reranker_model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"
    # Cap RAG retrieval so a hung/slow Chroma degrades to a typed 504 instead of
    # blocking every chat request indefinitely.
    rag_timeout_seconds: int = 10
    chunk_size: int = 512
    chunk_overlap: int = 64
    # Dataset synthesis: fail the run if more than this fraction of chunks errored,
    # rather than silently shipping a sparse/degraded dataset.
    synthesis_max_error_rate: float = 0.5

    # Auth brute-force rate limiting: fixed-window per-IP and per-email cap on
    # the unauthenticated auth endpoints (login/register/accept-invite). Fail-open: a
    # limiter (Redis) outage must never lock everyone out of auth.
    auth_rate_limit_max: int = 20  # allowed attempts per window per key
    auth_rate_limit_window_seconds: int = 60

    # Training — eval gate. An adapter is promotable if EITHER it clears the absolute
    # score floor (a strong fine-tune), OR it clears a low sanity floor AND meaningfully
    # beats the base model on the same held-out split (it demonstrably helped). The
    # improvement path matters because the absolute score is exp(-held_out_perplexity):
    # a small base model can't reach 0.6 even on an ideal task, yet a fine-tune that
    # reliably doubles the base score has clearly learned something.
    eval_score_threshold: float = 0.6  # absolute "strong adapter" pass
    eval_min_improvement: float = 0.05  # min score gain over base for the improvement path
    eval_min_floor: float = 0.05  # sanity floor for the improvement path (not garbage)
    # Minimum dataset size to start a training job. Below this the held-out
    # eval split collapses (e.g. 1 row → 0 held out → the gate scores the training
    # rows and only measures memorization), so we reject the job up front.
    min_training_samples: int = 10
    # Free-disk preflight for the worker: bail before training if the adapters
    # volume has less than this much free, rather than dying deep in a run.
    min_free_disk_gb: float = 5.0

    # Image dataset bundles (§V2): caps applied while extracting/validating a
    # .zip bundle (JSONL manifest + images). Outside these a bundle is rejected
    # at validation, never accepted into a run that would die mid-train.
    max_bundle_files: int = 2000
    max_bundle_uncompressed_mb: int = 500
    max_image_mb: int = 10
    max_image_side_px: int = 8192
    # Hard cap on a single dataset UPLOAD (the compressed .zip / .jsonl on the wire),
    # enforced while streaming to disk so a hostile/oversized upload can't fill the disk
    # before validation even runs. Distinct from max_bundle_uncompressed_mb (post-extract).
    max_upload_mb: int = 1024

    # Vision serving (§V4): caps on image content-parts per chat request.
    # Decoded size/dimensions reuse max_image_mb / max_image_side_px above.
    max_images_per_request: int = 4

    # Fine-tune serving (A3.1): PEFT adapters are converted to a GGUF LoRA so the
    # single llama-cpp runtime can serve them via `lora_path`. The converter is
    # llama.cpp's official convert_lora_to_gguf.py, vendored into the worker image.
    lora_convert_dir: Path = Path("/opt/llamacpp")  # holds convert_lora_to_gguf.py
    lora_gguf_filename: str = "adapter.gguf"  # converted artifact, stored beside safetensors
    lora_outtype: str = "f16"  # GGUF LoRA quant for conversion (f16/f32/q8_0)

    # Serving backend (D3): llama-cpp is the default and the only path for RAG/CPU
    # and VLM mmproj serving. Set to "vllm" to route text LoRA endpoints through
    # an external vLLM server (see OPERATIONS.md §9 and docker-compose.yml's
    # ``profiles: [vllm]``). vLLM serves N adapters from one GPU process via
    # continuous batching, closing the density gap confirmed in D0 Q3.
    serving_backend: str = "llamacpp"  # "llamacpp" (default) | "vllm"
    vllm_base_url: str = "http://vllm-server:8000"  # vllm-server service URL
    vllm_max_loras: int = 8  # max simultaneously-loaded LoRA adapters in vLLM

    # CORS
    cors_origins: list[str] = []  # empty = no CORS; override in production via ADAPTA_CORS_ORIGINS

    # Observability
    log_format: str = "text"  # "text" (default) or "json" for structured logs
    metrics_enabled: bool = True  # expose GET /metrics (Prometheus text format)

    @model_validator(mode="before")
    @classmethod
    def _validate_secret_key(cls, values: dict) -> dict:
        secret = values.get("secret_key", "CHANGE_ME_IN_PRODUCTION_use_openssl_rand_hex_32")
        if "CHANGE_ME" in str(secret):
            import warnings

            warnings.warn(
                "ADAPTA_SECRET_KEY is still the default placeholder. "
                "Set a strong secret with: openssl rand -hex 32",
                stacklevel=2,
            )
        return values

    @model_validator(mode="before")
    @classmethod
    def _derive_subdirs(cls, values: dict) -> dict:
        # If ADAPTA_DATA_DIR is set, re-base any subdir that wasn't explicitly
        # overridden. This runs before field parsing, so all annotations stay Path.
        data_dir = values.get("data_dir")
        if data_dir is None:
            return values
        base = Path(str(data_dir))
        for name, child in (
            ("models_dir", "models"),
            ("uploads_dir", "uploads"),
            ("adapters_dir", "adapters"),
            ("datasets_dir", "datasets"),
        ):
            if name not in values:
                values[name] = base / child
        return values

    def ensure_dirs(self) -> None:
        for d in [
            self.data_dir,
            self.models_dir,
            self.uploads_dir,
            self.adapters_dir,
            self.datasets_dir,
        ]:
            d.mkdir(parents=True, exist_ok=True)

---

Services — business logic

Chat (the serving path)

The single real serving path: RAG retrieval + context fitting + inference + usage metering. See the request lifecycle.

Single serving path for RAG and fine-tune (LoRA) endpoints.

Composes retrieval, context fitting, and inference. Does NOT modify adapta/core/inference.py or adapta/core/model_manager.py internals — those are called via their public interfaces only (hard constraint #1 in CLAUDE.md).

flatten_text

flatten_text(content) -> str

The text of a message whose content may be a string or a parts array. Used for RAG queries and text-path serving (image parts contribute nothing).

Source code in adapta/services/chat.py

def flatten_text(content) -> str:
    """The text of a message whose content may be a string or a parts array.
    Used for RAG queries and text-path serving (image parts contribute nothing)."""
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        return " ".join(
            p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text"
        ).strip()
    return ""

has_image_parts

has_image_parts(messages: List[dict]) -> bool

True if any message carries an image_url content-part. The router uses this BEFORE building a StreamingResponse — a typed error must be raised pre-stream, not mid-stream.

Source code in adapta/services/chat.py

def has_image_parts(messages: List[dict]) -> bool:
    """True if any message carries an image_url content-part. The router uses
    this BEFORE building a StreamingResponse — a typed error must be raised
    pre-stream, not mid-stream."""
    for m in messages:
        content = m.get("content")
        if isinstance(content, list):
            for p in content:
                if isinstance(p, dict) and p.get("type") == "image_url":
                    return True
    return False

validate_image_parts

validate_image_parts(messages: List[dict]) -> int

Validate every image content-part (count cap, data-URL form, base64, size, decodable image, dimension cap) and return the total estimated image token cost. Raises typed 422s — image input must never 500 (§V4.2).

Source code in adapta/services/chat.py

def validate_image_parts(messages: List[dict]) -> int:
    """Validate every image content-part (count cap, data-URL form, base64,
    size, decodable image, dimension cap) and return the total estimated image
    token cost. Raises typed 422s — image input must never 500 (§V4.2)."""
    import io

    from PIL import Image

    count = 0
    est_tokens = 0
    for m in messages:
        content = m.get("content")
        if not isinstance(content, list):
            continue
        for p in content:
            if not (isinstance(p, dict) and p.get("type") == "image_url"):
                continue
            count += 1
            if count > settings.max_images_per_request:
                raise InvalidRequest(
                    message=(
                        f"Too many images: at most {settings.max_images_per_request} "
                        "per request."
                    )
                )
            url = (p.get("image_url") or {}).get("url", "")
            raw = _decode_data_url(url)
            try:
                with Image.open(io.BytesIO(raw)) as img:
                    width, height = img.size
            except Exception as exc:
                raise InvalidRequest(
                    message="Image payload could not be decoded as an image."
                ) from exc
            if max(width, height) > settings.max_image_side_px:
                raise InvalidRequest(
                    message=(
                        f"Image side {max(width, height)}px exceeds the "
                        f"{settings.max_image_side_px}px limit."
                    )
                )
            est_tokens += _estimate_image_tokens(width, height)
    return est_tokens

chat async

chat(
    *,
    model_name: str,
    messages: List[dict],
    system_prompt: Optional[str] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    top_p: Optional[float] = None,
    stream: bool = False,
    project_id: Optional[str] = None,
    top_k_rag: Optional[int] = None,
    adapter_path: Optional[str] = None
) -> dict

Non-streaming chat completion. Returns an OpenAI-compatible response dict with optional citations.

Source code in adapta/services/chat.py

async def chat(
    *,
    model_name: str,
    messages: List[dict],
    system_prompt: Optional[str] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    top_p: Optional[float] = None,
    stream: bool = False,
    # RAG
    project_id: Optional[str] = None,
    top_k_rag: Optional[int] = None,
    # Fine-tune serving (A3.1): GGUF LoRA applied on top of the base model
    adapter_path: Optional[str] = None,
) -> dict:
    """
    Non-streaming chat completion.
    Returns an OpenAI-compatible response dict with optional citations.
    """
    temperature = temperature if temperature is not None else settings.temperature
    # Cap the client's max_tokens at the configured ceiling (A4.3) — an uncapped
    # request (e.g. max_tokens=999999) could OOM or hang the engine.
    max_tokens = min(max_tokens or settings.max_tokens, settings.max_tokens)
    top_p = top_p or settings.top_p

    # Image content-parts (§V4): only on a vision base, validated + capped, and
    # served through the chat handler. v1 design decision: requests carrying
    # images skip document retrieval (no RAG composition with image input yet) —
    # text-only requests on the same endpoint still compose RAG as usual.
    if has_image_parts(messages):
        if _base_modality(model_name) != "vision":
            raise InvalidRequest(
                message=(
                    "This endpoint's base model is text-only and cannot accept image "
                    "content. Create the project on a vision base model "
                    "(e.g. qwen2.5-vl-3b-instruct)."
                )
            )
        image_tokens = validate_image_parts(messages)
        return await _chat_vision(
            model_name=model_name,
            messages=messages,
            image_tokens=image_tokens,
            temperature=temperature,
            max_tokens=max_tokens,
            top_p=top_p,
            adapter_path=adapter_path,
        )

    rag_chunks: list = []
    rag_service = None
    if project_id:
        rag_service = get_rag_service()
        query = flatten_text(messages[-1].get("content", "")) if messages else ""
        rag_chunks = await _retrieve(
            rag_service, project_id, query, top_k_rag or settings.rag_top_k
        )

    # Content may be a parts array (all-text on the text path); flatten to the
    # plain string the manual prompt formatter expects.
    inference_messages = [
        Message(role=m["role"], content=flatten_text(m["content"])) for m in messages
    ]

    # Backend selection (D3): llama-cpp default; vLLM when ADAPTA_SERVING_BACKEND=vllm
    # and the request carries a text LoRA adapter. Vision path (_chat_vision) always
    # uses llama-cpp directly — vLLM does not support VLM LoRA layers.
    backend = get_backend(model_name, adapter_path)
    handle = await backend.prepare(model_name, adapter_path)

    # Fit prompt + answer into the model's context window before dispatch.
    # count_prompt_tokens / context_size are synchronous (llama-cpp: C call; vLLM:
    # char//4 estimate); run the whole fit off the event loop so a large prompt
    # never stalls other requests (e.g. /health).
    def _fit():
        return _fit_context(
            count_fn=lambda sp: handle.count_prompt_tokens(inference_messages, sp),
            n_ctx=handle.context_size(),
            base_system=system_prompt,
            rag_service=rag_service,
            rag_chunks=rag_chunks,
            max_tokens=max_tokens,
        )

    full_system, rag_chunks, max_tokens = await asyncio.to_thread(_fit)

    req = InferenceRequest(
        messages=inference_messages,
        model_name=model_name,
        temperature=temperature,
        top_p=top_p,
        max_tokens=max_tokens,
        stream=False,
        system_prompt=full_system,
        adapter_path=adapter_path,
    )
    try:
        response: InferenceResponse = await backend.generate(handle, req)
    except DomainError:
        raise  # Timeout (504) and other typed errors keep their status — don't mask as 500.
    except Exception as exc:
        raise InferenceFailed(
            message="Inference failed",
            internal_detail=str(exc),
        ) from exc

    result = {
        "id": f"chatcmpl-{uuid.uuid4().hex[:8]}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": model_name,
        "choices": [
            {
                "index": 0,
                "message": {"role": "assistant", "content": response.content},
                "finish_reason": response.finish_reason,
            }
        ],
        "usage": {
            "prompt_tokens": response.prompt_tokens,
            "completion_tokens": response.completion_tokens,
            "total_tokens": response.total_tokens,
        },
    }

    if rag_chunks:
        rag_service = get_rag_service()
        result["citations"] = rag_service.format_citations(rag_chunks)

    return result

chat_stream async

chat_stream(
    *,
    model_name: str,
    messages: List[dict],
    system_prompt: Optional[str] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    project_id: Optional[str] = None,
    top_k_rag: Optional[int] = None,
    endpoint_id: Optional[str] = None,
    adapter_path: Optional[str] = None
) -> AsyncIterator[str]

Streaming chat — yields SSE-formatted strings. RAG context is injected before streaming starts (non-streaming retrieval).

Source code in adapta/services/chat.py

async def chat_stream(
    *,
    model_name: str,
    messages: List[dict],
    system_prompt: Optional[str] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    project_id: Optional[str] = None,
    top_k_rag: Optional[int] = None,
    endpoint_id: Optional[str] = None,
    adapter_path: Optional[str] = None,
) -> AsyncIterator[str]:
    """
    Streaming chat — yields SSE-formatted strings.
    RAG context is injected before streaming starts (non-streaming retrieval).
    """
    import json

    temperature = temperature if temperature is not None else settings.temperature
    max_tokens = min(max_tokens or settings.max_tokens, settings.max_tokens)  # cap (A4.3)

    # Defensive (§V4): the router rejects stream+images BEFORE building the
    # StreamingResponse (a typed error here would surface mid-stream as a broken
    # body, not a 422). This guard only protects direct callers.
    if has_image_parts(messages):
        raise InvalidRequest(
            message="Streaming with image content is not supported; send stream=false."
        )

    rag_chunks: list = []
    rag_service = None
    if project_id:
        rag_service = get_rag_service()
        query = flatten_text(messages[-1].get("content", "")) if messages else ""
        rag_chunks = await _retrieve(
            rag_service, project_id, query, top_k_rag or settings.rag_top_k
        )

    inference_messages = [
        Message(role=m["role"], content=flatten_text(m["content"])) for m in messages
    ]

    # Backend selection (D3): same routing as non-streaming chat().
    backend = get_backend(model_name, adapter_path)
    handle = await backend.prepare(model_name, adapter_path)

    # Fit prompt + answer into the context window before streaming (A4.3).
    full_system, rag_chunks, max_tokens = _fit_context(
        count_fn=lambda sp: handle.count_prompt_tokens(inference_messages, sp),
        n_ctx=handle.context_size(),
        base_system=system_prompt,
        rag_service=rag_service,
        rag_chunks=rag_chunks,
        max_tokens=max_tokens,
    )

    req = InferenceRequest(
        messages=inference_messages,
        model_name=model_name,
        temperature=temperature,
        max_tokens=max_tokens,
        stream=True,
        system_prompt=full_system,
        adapter_path=adapter_path,
    )
    chunk_id = f"chatcmpl-{uuid.uuid4().hex[:8]}"
    created = int(time.time())

    # llama-cpp: exact tokenizer count.  vLLM: char//4 approximation (no tokenizer RPC).
    prompt_tokens = handle.count_prompt_tokens(inference_messages, full_system)
    # Each iteration is one generated token, so counting iterations is exact.
    completion_tokens = 0
    try:
        async for token in backend.generate_stream(handle, req):
            completion_tokens += 1
            chunk = {
                "id": chunk_id,
                "object": "chat.completion.chunk",
                "created": created,
                "model": model_name,
                "choices": [{"index": 0, "delta": {"content": token}, "finish_reason": None}],
            }
            yield f"data: {json.dumps(chunk)}\n\n"

        finish = {
            "id": chunk_id,
            "object": "chat.completion.chunk",
            "created": created,
            "model": model_name,
            "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
            "usage": {
                "prompt_tokens": prompt_tokens,
                "completion_tokens": completion_tokens,
                "total_tokens": prompt_tokens + completion_tokens,
            },
        }
        yield f"data: {json.dumps(finish)}\n\n"
        yield "data: [DONE]\n\n"

        # Metered off the client path — runs after the last byte is yielded.
        if endpoint_id:
            from adapta.services.usage import record_usage

            await record_usage(endpoint_id, prompt_tokens, completion_tokens)

    except DomainError:
        raise  # Timeout (504) and other typed errors keep their status — don't mask as 500.
    except Exception as exc:
        raise InferenceFailed(
            message="Streaming inference failed", internal_detail=str(exc)
        ) from exc

RAG (retrieval)

RAG service: per-project ChromaDB collections + hybrid cited retrieval.

Retrieval pipeline (D5): 1. Vector search — fetch top_k × multiplier candidates from Chroma. 2. BM25 score — keyword signal over the same candidates. 3. Reciprocal Rank Fusion — merge the two ranked lists. 4. Optional cross-encoder rerank over the top candidates. 5. Return top_k chunks, best-first (§A4.3 pop() invariant preserved).

RAGService

Per-project ChromaDB collection management and hybrid cited retrieval.

A new ChromaDB HTTP client is created on every call (_client() → _get_chroma_client()); there is no persistent connection held by this object.

Source code in adapta/services/rag.py

class RAGService:
    """Per-project ChromaDB collection management and hybrid cited retrieval.

    A new ChromaDB HTTP client is created on every call (``_client()`` →
    ``_get_chroma_client()``); there is no persistent connection held by
    this object.
    """

    def _client(self):
        return _get_chroma_client()

    def ensure_collection(self, project_id: str) -> str:
        """Create the Chroma collection if it doesn't exist; returns the name.

        Sets ``hnsw:space: cosine`` so the index uses cosine distance.  This
        metadata is applied only at creation — Chroma ignores it on
        ``get_or_create_collection`` calls for an existing collection, so the
        distance metric is locked in after the first call.
        """
        name = collection_name_for(project_id)
        client = self._client()
        client.get_or_create_collection(name=name, metadata={"hnsw:space": "cosine"})
        return name

    def delete_collection(self, project_id: str) -> None:
        """Best-effort cleanup; silently swallows all errors.

        Called on project delete — a Chroma outage or already-absent collection
        must not block the project row from being removed from Postgres.
        """
        name = collection_name_for(project_id)
        client = self._client()
        try:
            client.delete_collection(name)
        except Exception:
            pass

    def index_chunks(
        self,
        project_id: str,
        file_id: str,
        chunks: list,  # List[Chunk] from documents.py
    ) -> int:
        """Embed and upsert chunks into the project collection; returns count.

        Uses ``upsert`` so re-indexing a file is idempotent — existing chunks
        for the same ``file_id`` are overwritten rather than duplicated.
        """
        if not chunks:
            return 0

        emb_service = get_embedding_service()
        texts = [c.text for c in chunks]
        embeddings = emb_service.embed(texts)

        name = collection_name_for(project_id)
        client = self._client()
        collection = client.get_or_create_collection(name=name, metadata={"hnsw:space": "cosine"})

        ids = [f"{file_id}_{c.index}" for c in chunks]
        metadatas = [
            {"source": c.source, "chunk_index": c.index, "file_id": file_id} for c in chunks
        ]

        collection.upsert(ids=ids, documents=texts, embeddings=embeddings, metadatas=metadatas)
        return len(chunks)

    def delete_file_chunks(self, project_id: str, file_id: str) -> None:
        """Remove all chunks for a file from the Chroma collection.

        Two-step (``get`` then ``delete``) because Chroma's ``delete`` requires
        explicit IDs rather than accepting a ``where`` filter directly.
        """
        name = collection_name_for(project_id)
        client = self._client()
        try:
            collection = client.get_collection(name)
            results = collection.get(where={"file_id": file_id})
            if results["ids"]:
                collection.delete(ids=results["ids"])
        except Exception as exc:
            logger.warning("Failed to delete file chunks: %s", exc)

    def retrieve(
        self,
        project_id: str,
        query: str,
        top_k: Optional[int] = None,
    ) -> List[RetrievedChunk]:
        """Hybrid retrieval: vector + BM25 fused via RRF, optional cross-encoder reranker.

        Pipeline (D5):
          1. Fetch fetch_k = top_k × multiplier candidates via Chroma vector search.
          2. BM25-score those candidates against the query.
          3. Reciprocal Rank Fusion of vector order + BM25 order.
          4. Optional cross-encoder reranker over the top rerank_n candidates.
          5. Return top_k, best-first (§A4.3 pop() invariant preserved).

        Citations reflect the final reranked scores (§A4.3 ordering).
        """
        top_k = top_k or settings.rag_top_k
        fetch_k = max(top_k * settings.rag_hybrid_fetch_multiplier, top_k + 1)

        emb_service = get_embedding_service()
        query_embedding = emb_service.embed_one(query)

        name = collection_name_for(project_id)
        client = self._client()
        try:
            collection = client.get_collection(name)
        except Exception:
            return []

        results = collection.query(
            query_embeddings=[query_embedding],
            n_results=fetch_k,
            include=["documents", "metadatas", "distances"],
        )

        docs: list[str] = results["documents"][0]
        metas: list[dict] = results["metadatas"][0]
        dists: list[float] = results["distances"][0]

        if not docs:
            return []

        bm25 = _bm25_scores(query, docs)
        vec_order = list(range(len(docs)))
        bm25_order = sorted(range(len(docs)), key=lambda i: bm25[i], reverse=True)
        fused_order = _rrf_fuse(vec_order, bm25_order)

        candidates: List[RetrievedChunk] = [
            RetrievedChunk(
                text=docs[idx],
                source=metas[idx].get("source", "unknown"),
                score=1.0 - dists[idx],  # cosine distance → similarity
                chunk_index=metas[idx].get("chunk_index", 0),
            )
            for idx in fused_order
        ]

        reranker = get_reranker_service()
        if reranker is not None and len(candidates) > 1:
            rerank_n = min(top_k * 2, len(candidates))
            to_rerank = candidates[:rerank_n]
            raw_scores = reranker.rerank(query, [c.text for c in to_rerank])
            ranked = sorted(
                zip(raw_scores, to_rerank, strict=False),
                key=lambda x: x[0],
                reverse=True,
            )
            # Cross-encoder logits are unbounded; sigmoid maps them to (0, 1)
            # so the citation `score` field is comparable across requests.
            reranked = [
                RetrievedChunk(
                    text=c.text,
                    source=c.source,
                    score=round(1.0 / (1.0 + math.exp(-s)), 4),
                    chunk_index=c.chunk_index,
                )
                for s, c in ranked
            ]
            candidates = reranked + candidates[rerank_n:]

        return candidates[:top_k]

    def build_context_block(self, chunks: List[RetrievedChunk]) -> str:
        """Format retrieved chunks into a numbered context block for the system prompt.

        Produces the ``[N] (source: X)\\ntext`` format that the system-prompt
        template in ``chat.py`` references with "Cite sources by their [N]
        number."  The numbers here must stay consistent with the indices in
        ``format_citations`` — both iterate in the same order.
        """
        if not chunks:
            return ""
        parts = []
        for i, c in enumerate(chunks, 1):
            parts.append(f"[{i}] (source: {c.source})\n{c.text}")
        return "\n\n---\n\n".join(parts)

    def format_citations(self, chunks: List[RetrievedChunk]) -> list[dict]:
        return [
            {"index": i + 1, "source": c.source, "score": round(c.score, 4)}
            for i, c in enumerate(chunks)
        ]

ensure_collection

ensure_collection(project_id: str) -> str

Create the Chroma collection if it doesn't exist; returns the name.

Sets hnsw:space: cosine so the index uses cosine distance. This metadata is applied only at creation — Chroma ignores it on get_or_create_collection calls for an existing collection, so the distance metric is locked in after the first call.

Source code in adapta/services/rag.py

def ensure_collection(self, project_id: str) -> str:
    """Create the Chroma collection if it doesn't exist; returns the name.

    Sets ``hnsw:space: cosine`` so the index uses cosine distance.  This
    metadata is applied only at creation — Chroma ignores it on
    ``get_or_create_collection`` calls for an existing collection, so the
    distance metric is locked in after the first call.
    """
    name = collection_name_for(project_id)
    client = self._client()
    client.get_or_create_collection(name=name, metadata={"hnsw:space": "cosine"})
    return name

delete_collection

delete_collection(project_id: str) -> None

Best-effort cleanup; silently swallows all errors.

Called on project delete — a Chroma outage or already-absent collection must not block the project row from being removed from Postgres.

Source code in adapta/services/rag.py

def delete_collection(self, project_id: str) -> None:
    """Best-effort cleanup; silently swallows all errors.

    Called on project delete — a Chroma outage or already-absent collection
    must not block the project row from being removed from Postgres.
    """
    name = collection_name_for(project_id)
    client = self._client()
    try:
        client.delete_collection(name)
    except Exception:
        pass

index_chunks

index_chunks(
    project_id: str, file_id: str, chunks: list
) -> int

Embed and upsert chunks into the project collection; returns count.

Uses upsert so re-indexing a file is idempotent — existing chunks for the same file_id are overwritten rather than duplicated.

Source code in adapta/services/rag.py

def index_chunks(
    self,
    project_id: str,
    file_id: str,
    chunks: list,  # List[Chunk] from documents.py
) -> int:
    """Embed and upsert chunks into the project collection; returns count.

    Uses ``upsert`` so re-indexing a file is idempotent — existing chunks
    for the same ``file_id`` are overwritten rather than duplicated.
    """
    if not chunks:
        return 0

    emb_service = get_embedding_service()
    texts = [c.text for c in chunks]
    embeddings = emb_service.embed(texts)

    name = collection_name_for(project_id)
    client = self._client()
    collection = client.get_or_create_collection(name=name, metadata={"hnsw:space": "cosine"})

    ids = [f"{file_id}_{c.index}" for c in chunks]
    metadatas = [
        {"source": c.source, "chunk_index": c.index, "file_id": file_id} for c in chunks
    ]

    collection.upsert(ids=ids, documents=texts, embeddings=embeddings, metadatas=metadatas)
    return len(chunks)

delete_file_chunks

delete_file_chunks(project_id: str, file_id: str) -> None

Remove all chunks for a file from the Chroma collection.

Two-step (get then delete) because Chroma's delete requires explicit IDs rather than accepting a where filter directly.

Source code in adapta/services/rag.py

def delete_file_chunks(self, project_id: str, file_id: str) -> None:
    """Remove all chunks for a file from the Chroma collection.

    Two-step (``get`` then ``delete``) because Chroma's ``delete`` requires
    explicit IDs rather than accepting a ``where`` filter directly.
    """
    name = collection_name_for(project_id)
    client = self._client()
    try:
        collection = client.get_collection(name)
        results = collection.get(where={"file_id": file_id})
        if results["ids"]:
            collection.delete(ids=results["ids"])
    except Exception as exc:
        logger.warning("Failed to delete file chunks: %s", exc)

retrieve

retrieve(
    project_id: str, query: str, top_k: Optional[int] = None
) -> List[RetrievedChunk]

Hybrid retrieval: vector + BM25 fused via RRF, optional cross-encoder reranker.

Pipeline (D5): 1. Fetch fetch_k = top_k × multiplier candidates via Chroma vector search. 2. BM25-score those candidates against the query. 3. Reciprocal Rank Fusion of vector order + BM25 order. 4. Optional cross-encoder reranker over the top rerank_n candidates. 5. Return top_k, best-first (§A4.3 pop() invariant preserved).

Citations reflect the final reranked scores (§A4.3 ordering).

Source code in adapta/services/rag.py

def retrieve(
    self,
    project_id: str,
    query: str,
    top_k: Optional[int] = None,
) -> List[RetrievedChunk]:
    """Hybrid retrieval: vector + BM25 fused via RRF, optional cross-encoder reranker.

    Pipeline (D5):
      1. Fetch fetch_k = top_k × multiplier candidates via Chroma vector search.
      2. BM25-score those candidates against the query.
      3. Reciprocal Rank Fusion of vector order + BM25 order.
      4. Optional cross-encoder reranker over the top rerank_n candidates.
      5. Return top_k, best-first (§A4.3 pop() invariant preserved).

    Citations reflect the final reranked scores (§A4.3 ordering).
    """
    top_k = top_k or settings.rag_top_k
    fetch_k = max(top_k * settings.rag_hybrid_fetch_multiplier, top_k + 1)

    emb_service = get_embedding_service()
    query_embedding = emb_service.embed_one(query)

    name = collection_name_for(project_id)
    client = self._client()
    try:
        collection = client.get_collection(name)
    except Exception:
        return []

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=fetch_k,
        include=["documents", "metadatas", "distances"],
    )

    docs: list[str] = results["documents"][0]
    metas: list[dict] = results["metadatas"][0]
    dists: list[float] = results["distances"][0]

    if not docs:
        return []

    bm25 = _bm25_scores(query, docs)
    vec_order = list(range(len(docs)))
    bm25_order = sorted(range(len(docs)), key=lambda i: bm25[i], reverse=True)
    fused_order = _rrf_fuse(vec_order, bm25_order)

    candidates: List[RetrievedChunk] = [
        RetrievedChunk(
            text=docs[idx],
            source=metas[idx].get("source", "unknown"),
            score=1.0 - dists[idx],  # cosine distance → similarity
            chunk_index=metas[idx].get("chunk_index", 0),
        )
        for idx in fused_order
    ]

    reranker = get_reranker_service()
    if reranker is not None and len(candidates) > 1:
        rerank_n = min(top_k * 2, len(candidates))
        to_rerank = candidates[:rerank_n]
        raw_scores = reranker.rerank(query, [c.text for c in to_rerank])
        ranked = sorted(
            zip(raw_scores, to_rerank, strict=False),
            key=lambda x: x[0],
            reverse=True,
        )
        # Cross-encoder logits are unbounded; sigmoid maps them to (0, 1)
        # so the citation `score` field is comparable across requests.
        reranked = [
            RetrievedChunk(
                text=c.text,
                source=c.source,
                score=round(1.0 / (1.0 + math.exp(-s)), 4),
                chunk_index=c.chunk_index,
            )
            for s, c in ranked
        ]
        candidates = reranked + candidates[rerank_n:]

    return candidates[:top_k]

build_context_block

build_context_block(chunks: List[RetrievedChunk]) -> str

Format retrieved chunks into a numbered context block for the system prompt.

Produces the [N] (source: X)\ntext format that the system-prompt template in chat.py references with "Cite sources by their [N] number." The numbers here must stay consistent with the indices in format_citations — both iterate in the same order.

Source code in adapta/services/rag.py

def build_context_block(self, chunks: List[RetrievedChunk]) -> str:
    """Format retrieved chunks into a numbered context block for the system prompt.

    Produces the ``[N] (source: X)\\ntext`` format that the system-prompt
    template in ``chat.py`` references with "Cite sources by their [N]
    number."  The numbers here must stay consistent with the indices in
    ``format_citations`` — both iterate in the same order.
    """
    if not chunks:
        return ""
    parts = []
    for i, c in enumerate(chunks, 1):
        parts.append(f"[{i}] (source: {c.source})\n{c.text}")
    return "\n\n---\n\n".join(parts)

collection_name_for

collection_name_for(project_id: str) -> str

Return the Chroma collection name for a project.

The convention proj_{project_id} is the value stored in Collection.chroma_collection_name. Changing this function without a migration would orphan all existing collections.

Source code in adapta/services/rag.py

def collection_name_for(project_id: str) -> str:
    """Return the Chroma collection name for a project.

    The convention ``proj_{project_id}`` is the value stored in
    ``Collection.chroma_collection_name``.  Changing this function without a
    migration would orphan all existing collections.
    """
    return f"proj_{project_id}"

get_rag_service

get_rag_service() -> RAGService

Return the process-level RAGService singleton.

Initialization is not protected by a lock. In asyncio this is safe because coroutines yield at await points — the first call sets _rag_service before any other coroutine reaches this function.

Source code in adapta/services/rag.py

def get_rag_service() -> RAGService:
    """Return the process-level RAGService singleton.

    Initialization is not protected by a lock.  In asyncio this is safe
    because coroutines yield at ``await`` points — the first call sets
    ``_rag_service`` before any other coroutine reaches this function.
    """
    global _rag_service
    if _rag_service is None:
        _rag_service = RAGService()
    return _rag_service

Embeddings

sentence-transformers embedding and cross-encoder reranking services.

Both services are lazily loaded on first use so the model weights are not downloaded at import time — only when an actual embed/rerank call is made. Both are cached as process-level singletons via lru_cache; the first call's model name is locked in for the lifetime of the process, so a config change requires a restart.

EmbeddingService

Bi-encoder sentence-transformers model, lazily loaded on first use.

Not thread-safe at load time: if two coroutines race the first call before the model is loaded, both see _model is None and both load. In practice the asyncio event loop serializes the first call, but the double-load is harmless (the second assignment overwrites with an identical model object).

Source code in adapta/services/embeddings.py

class EmbeddingService:
    """Bi-encoder sentence-transformers model, lazily loaded on first use.

    Not thread-safe at load time: if two coroutines race the first call
    before the model is loaded, both see ``_model is None`` and both load.
    In practice the asyncio event loop serializes the first call, but the
    double-load is harmless (the second assignment overwrites with an
    identical model object).
    """

    def __init__(self, model_name: str):
        self._model_name = model_name
        self._model = None

    def _load(self) -> None:
        if self._model is not None:
            return
        try:
            from sentence_transformers import SentenceTransformer

            logger.info("Loading embedding model: %s", self._model_name)
            self._model = SentenceTransformer(self._model_name)
            logger.info("Embedding model loaded")
        except Exception as exc:
            raise EmbeddingFailed(
                message="Failed to load embedding model",
                internal_detail=str(exc),
            ) from exc

    def embed(self, texts: List[str]) -> List[List[float]]:
        """Embed a batch of strings.  Returns one float vector per input."""
        if not texts:
            return []
        self._load()
        assert self._model is not None
        try:
            vectors = self._model.encode(texts, show_progress_bar=False)
            return [v.tolist() for v in vectors]
        except Exception as exc:
            raise EmbeddingFailed(
                message="Embedding generation failed",
                internal_detail=str(exc),
            ) from exc

    def embed_one(self, text: str) -> List[float]:
        """Embed a single string.  Equivalent to ``embed([text])[0]``.

        Not a separate model call — just a convenience wrapper around ``embed``.
        Use ``embed`` directly for batches to avoid per-call overhead.
        """
        return self.embed([text])[0]

    @property
    def dimension(self) -> int:
        """Embedding vector dimension.  Triggers model load on first access."""
        self._load()
        assert self._model is not None
        return self._model.get_sentence_embedding_dimension()  # type: ignore[return-value]

dimension property

dimension: int

Embedding vector dimension. Triggers model load on first access.

embed

embed(texts: List[str]) -> List[List[float]]

Embed a batch of strings. Returns one float vector per input.

Source code in adapta/services/embeddings.py

def embed(self, texts: List[str]) -> List[List[float]]:
    """Embed a batch of strings.  Returns one float vector per input."""
    if not texts:
        return []
    self._load()
    assert self._model is not None
    try:
        vectors = self._model.encode(texts, show_progress_bar=False)
        return [v.tolist() for v in vectors]
    except Exception as exc:
        raise EmbeddingFailed(
            message="Embedding generation failed",
            internal_detail=str(exc),
        ) from exc

embed_one

embed_one(text: str) -> List[float]

Embed a single string. Equivalent to embed([text])[0].

Not a separate model call — just a convenience wrapper around embed. Use embed directly for batches to avoid per-call overhead.

Source code in adapta/services/embeddings.py

def embed_one(self, text: str) -> List[float]:
    """Embed a single string.  Equivalent to ``embed([text])[0]``.

    Not a separate model call — just a convenience wrapper around ``embed``.
    Use ``embed`` directly for batches to avoid per-call overhead.
    """
    return self.embed([text])[0]

RerankerService

Cross-encoder reranker, lazily loaded on first use (D5).

A cross-encoder takes (query, passage) pairs and scores them jointly — unlike bi-encoders (EmbeddingService) which encode query and passage independently. Scores are raw logits (unbounded, may be negative) and are not comparable to cosine-similarity scores; callers that expose them should normalize first (e.g. via sigmoid — see rag.py).

Source code in adapta/services/embeddings.py

class RerankerService:
    """Cross-encoder reranker, lazily loaded on first use (D5).

    A cross-encoder takes ``(query, passage)`` pairs and scores them jointly —
    unlike bi-encoders (EmbeddingService) which encode query and passage
    independently.  Scores are raw logits (unbounded, may be negative) and are
    not comparable to cosine-similarity scores; callers that expose them should
    normalize first (e.g. via sigmoid — see ``rag.py``).
    """

    def __init__(self, model_name: str):
        self._model_name = model_name
        self._model = None

    def _load(self) -> None:
        if self._model is not None:
            return
        try:
            from sentence_transformers import CrossEncoder

            logger.info("Loading reranker model: %s", self._model_name)
            self._model = CrossEncoder(self._model_name)
            logger.info("Reranker model loaded")
        except Exception as exc:
            raise EmbeddingFailed(
                message="Failed to load reranker model",
                internal_detail=str(exc),
            ) from exc

    def rerank(self, query: str, passages: List[str]) -> List[float]:
        """Score each passage against the query.  Returns raw logits (higher = more relevant).

        Scores are unbounded floats from the cross-encoder — NOT cosine
        similarities.  Apply ``sigmoid`` to normalize to (0, 1) before
        exposing them in API responses.
        """
        if not passages:
            return []
        self._load()
        assert self._model is not None
        try:
            pairs = [(query, p) for p in passages]
            scores = self._model.predict(pairs, show_progress_bar=False)
            return [float(s) for s in scores]
        except Exception as exc:
            raise EmbeddingFailed(
                message="Reranking failed",
                internal_detail=str(exc),
            ) from exc

rerank

rerank(query: str, passages: List[str]) -> List[float]

Score each passage against the query. Returns raw logits (higher = more relevant).

Scores are unbounded floats from the cross-encoder — NOT cosine similarities. Apply sigmoid to normalize to (0, 1) before exposing them in API responses.

Source code in adapta/services/embeddings.py

def rerank(self, query: str, passages: List[str]) -> List[float]:
    """Score each passage against the query.  Returns raw logits (higher = more relevant).

    Scores are unbounded floats from the cross-encoder — NOT cosine
    similarities.  Apply ``sigmoid`` to normalize to (0, 1) before
    exposing them in API responses.
    """
    if not passages:
        return []
    self._load()
    assert self._model is not None
    try:
        pairs = [(query, p) for p in passages]
        scores = self._model.predict(pairs, show_progress_bar=False)
        return [float(s) for s in scores]
    except Exception as exc:
        raise EmbeddingFailed(
            message="Reranking failed",
            internal_detail=str(exc),
        ) from exc

get_embedding_service cached

get_embedding_service() -> EmbeddingService

Return the process-level EmbeddingService singleton.

lru_cache(maxsize=1) means the first call determines the model name for the process lifetime; subsequent calls return the cached instance regardless of any config change. Restart the process to pick up a new ADAPTA_EMBEDDING_MODEL.

Source code in adapta/services/embeddings.py

@lru_cache(maxsize=1)
def get_embedding_service() -> EmbeddingService:
    """Return the process-level EmbeddingService singleton.

    ``lru_cache(maxsize=1)`` means the first call determines the model name
    for the process lifetime; subsequent calls return the cached instance
    regardless of any config change.  Restart the process to pick up a new
    ``ADAPTA_EMBEDDING_MODEL``.
    """
    return EmbeddingService(settings.embedding_model)

get_reranker_service cached

get_reranker_service() -> Optional[RerankerService]

Return the process-level RerankerService singleton, or None if disabled.

Returns None when ADAPTA_RAG_RERANKER_MODEL is empty — this disables reranking in rag.py and falls back to RRF-fused vector+BM25 order. Same lru_cache process-lifetime lock-in as get_embedding_service.

Source code in adapta/services/embeddings.py

@lru_cache(maxsize=1)
def get_reranker_service() -> Optional[RerankerService]:
    """Return the process-level RerankerService singleton, or None if disabled.

    Returns None when ``ADAPTA_RAG_RERANKER_MODEL`` is empty — this disables
    reranking in ``rag.py`` and falls back to RRF-fused vector+BM25 order.
    Same ``lru_cache`` process-lifetime lock-in as ``get_embedding_service``.
    """
    model = settings.rag_reranker_model
    if not model:
        return None
    return RerankerService(model)

Training (validation & enqueue)

Training coordination service. Handles dataset validation and job enqueuing; actual training runs in the worker.

BundleError

Bases: ValueError

A dataset .zip bundle that cannot be safely extracted (zip-slip, caps, disallowed content). The message is operator-safe.

Source code in adapta/services/training.py

class BundleError(ValueError):
    """A dataset .zip bundle that cannot be safely extracted (zip-slip, caps,
    disallowed content). The message is operator-safe."""

extract_bundle

extract_bundle(zip_path: Path, dest_dir: Path) -> Path

Safely extract a dataset bundle (§V2.1) and return the manifest path.

A bundle is a .zip holding exactly one *.jsonl manifest at its root and the images the manifest references (ALLOWED_IMAGE_EXTENSIONS). Hostile archives are rejected before any byte is written: path traversal (zip-slip), absolute paths, symlinks, over-cap file counts / uncompressed size, and disallowed file types all raise :class:BundleError.

Source code in adapta/services/training.py

def extract_bundle(zip_path: Path, dest_dir: Path) -> Path:
    """Safely extract a dataset bundle (§V2.1) and return the manifest path.

    A bundle is a .zip holding exactly one ``*.jsonl`` manifest at its root and
    the images the manifest references (``ALLOWED_IMAGE_EXTENSIONS``). Hostile
    archives are rejected before any byte is written: path traversal (zip-slip),
    absolute paths, symlinks, over-cap file counts / uncompressed size, and
    disallowed file types all raise :class:`BundleError`.
    """
    import zipfile

    if not zipfile.is_zipfile(zip_path):
        raise BundleError("Not a valid .zip archive")

    max_bytes = settings.max_bundle_uncompressed_mb * 1024 * 1024
    dest_dir = dest_dir.resolve()

    with zipfile.ZipFile(zip_path) as zf:
        members = [m for m in zf.infolist() if not m.is_dir()]
        if len(members) > settings.max_bundle_files:
            raise BundleError(
                f"Bundle has {len(members)} files; at most {settings.max_bundle_files} are allowed"
            )
        total = sum(m.file_size for m in members)
        if total > max_bytes:
            raise BundleError(
                f"Bundle uncompressed size ({total // (1024 * 1024)} MB) exceeds the "
                f"{settings.max_bundle_uncompressed_mb} MB limit"
            )

        manifests: list[str] = []
        for m in members:
            name = m.filename
            # Zip-slip / absolute-path / drive-letter protection: the resolved
            # destination must stay inside dest_dir.
            if (
                name.startswith(("/", "\\"))
                or ".." in Path(name).parts
                or ":" in name.split("/")[0]
            ):
                raise BundleError(f"Unsafe path in bundle: {name!r}")
            resolved = (dest_dir / name).resolve()
            if not resolved.is_relative_to(dest_dir):
                raise BundleError(f"Unsafe path in bundle: {name!r}")
            # Symlink entries (external_attr high bits = S_IFLNK) would let a
            # later member write through the link — reject outright.
            if (m.external_attr >> 16) & 0o170000 == 0o120000:
                raise BundleError(f"Symlinks are not allowed in bundles: {name!r}")
            ext = Path(name).suffix.lower()
            if ext == ".jsonl":
                if "/" in name:
                    raise BundleError("The .jsonl manifest must be at the bundle root")
                manifests.append(name)
            elif ext not in ALLOWED_IMAGE_EXTENSIONS:
                raise BundleError(
                    f"Disallowed file type in bundle: {name!r} (allowed: one root .jsonl + "
                    + "/".join(sorted(e.lstrip(".") for e in ALLOWED_IMAGE_EXTENSIONS))
                    + ")"
                )

        if len(manifests) != 1:
            raise BundleError(
                f"Bundle must contain exactly one root .jsonl manifest (found {len(manifests)})"
            )

        # Extract member-by-member with a HARD cap on actual bytes written. The earlier
        # ``sum(m.file_size)`` pre-check trusts the zip's central directory, which an
        # attacker controls — a "zip bomb" can declare a small uncompressed size yet
        # decompress to gigabytes. Streaming each entry and aborting once the real total
        # crosses the cap is what actually bounds disk use.
        dest_dir.mkdir(parents=True, exist_ok=True)
        written = 0
        for m in members:
            target = (dest_dir / m.filename).resolve()
            target.parent.mkdir(parents=True, exist_ok=True)
            with zf.open(m) as src, target.open("wb") as out:
                while True:
                    chunk = src.read(1024 * 1024)
                    if not chunk:
                        break
                    written += len(chunk)
                    if written > max_bytes:
                        raise BundleError(
                            "Bundle decompresses beyond the "
                            f"{settings.max_bundle_uncompressed_mb} MB limit"
                        )
                    out.write(chunk)

    return dest_dir / manifests[0]

validate_dataset

validate_dataset(
    path: Path, bundle_dir: Optional[Path] = None
) -> tuple[bool, Optional[str], int, int]

Validate a JSONL file against the training dataset schema.

Returns (is_valid, error_message, num_samples, num_images).

Accepts both SFT rows (prompt + response) and DPO preference rows (prompt + chosen + rejected). Rows within a single dataset must all be the same type — mixing SFT and DPO rows is rejected.

Validation does not stop at the first bad row: it scans the whole file and reports up to :data:MAX_REPORTED_ERRORS problems at once, so an operator fixing a large bundle sees every issue in one pass instead of discovering them one re-upload at a time.

Source code in adapta/services/training.py

def validate_dataset(
    path: Path, bundle_dir: Optional[Path] = None
) -> tuple[bool, Optional[str], int, int]:
    """Validate a JSONL file against the training dataset schema.

    Returns (is_valid, error_message, num_samples, num_images).

    Accepts both SFT rows (prompt + response) and DPO preference rows
    (prompt + chosen + rejected). Rows within a single dataset must all be
    the same type — mixing SFT and DPO rows is rejected.

    Validation does **not** stop at the first bad row: it scans the whole file
    and reports up to :data:`MAX_REPORTED_ERRORS` problems at once, so an
    operator fixing a large bundle sees every issue in one pass instead of
    discovering them one re-upload at a time.
    """
    try:
        schema = _load_schema()
        num_samples = 0
        num_images = 0
        errors: list[str] = []
        detected_methods: set[str] = set()
        with path.open(encoding="utf-8") as f:
            for i, line in enumerate(f, 1):
                line = line.strip()
                if not line:
                    continue
                try:
                    obj = json.loads(line)
                except json.JSONDecodeError as exc:
                    errors.append(f"Line {i}: invalid JSON — {exc}")
                    if len(errors) >= MAX_REPORTED_ERRORS:
                        break
                    continue

                row_errors, counted, row_method = _validate_row(i, obj, schema, bundle_dir)
                detected_methods.add(row_method)
                if row_errors:
                    errors.extend(row_errors)
                    if len(errors) >= MAX_REPORTED_ERRORS:
                        break
                    continue

                num_samples += 1
                num_images += counted

        if len(detected_methods) > 1:
            errors.append(
                "Dataset mixes SFT rows (prompt/response) and DPO rows "
                "(prompt/chosen/rejected) — a dataset must be one type only."
            )

        if errors:
            return False, _format_error_report(errors), 0, 0
        if num_samples == 0:
            return False, "Dataset is empty — must have at least one row", 0, 0

        return True, None, num_samples, num_images

    except Exception as exc:
        return False, str(exc), 0, 0

update_job_record async

update_job_record(
    job_id: str,
    *,
    status: Optional[str] = None,
    progress: Optional[float] = None,
    logs: Optional[str] = None,
    adapter_path: Optional[str] = None,
    eval_score: Optional[float] = None,
    eval_passed: Optional[bool] = None,
    eval_metrics: Optional[str] = None,
    error: Optional[str] = None
) -> None

Persist a training job's lifecycle to Postgres (the source of truth the API and the endpoint-creation gate read).

The Redis JobQueue.update_status only updates live progress; without this the Postgres row stayed at queued forever, so a finished job never looked terminal to GET /jobs/{id} and a passed adapter could never bind an endpoint.

Source code in adapta/services/training.py

async def update_job_record(
    job_id: str,
    *,
    status: Optional[str] = None,
    progress: Optional[float] = None,
    logs: Optional[str] = None,
    adapter_path: Optional[str] = None,
    eval_score: Optional[float] = None,
    eval_passed: Optional[bool] = None,
    eval_metrics: Optional[str] = None,
    error: Optional[str] = None,
) -> None:
    """Persist a training job's lifecycle to Postgres (the source of truth the API
    and the endpoint-creation gate read).

    The Redis ``JobQueue.update_status`` only updates live progress; without this the
    Postgres row stayed at ``queued`` forever, so a finished job never looked terminal
    to ``GET /jobs/{id}`` and a passed adapter could never bind an endpoint.
    """
    from datetime import datetime, timezone

    from sqlalchemy import select

    from adapta.db.session import AsyncSessionLocal

    async with AsyncSessionLocal() as db:
        row = (
            await db.execute(select(TrainingJob).where(TrainingJob.id == job_id))
        ).scalar_one_or_none()
        if row is None:
            logger.error("update_job_record: job %s not found — cannot persist status", job_id)
            return
        if status is not None:
            row.status = JobStatus(status)
            if status == "running" and row.started_at is None:
                row.started_at = datetime.now(timezone.utc)
            if status in ("succeeded", "failed", "cancelled"):
                row.finished_at = datetime.now(timezone.utc)
        if progress is not None:
            row.progress = progress
        if logs is not None:
            row.logs = logs
        if adapter_path is not None:
            row.adapter_path = adapter_path
        if eval_score is not None:
            row.eval_score = eval_score
        if eval_passed is not None:
            row.eval_passed = eval_passed
        if eval_metrics is not None:
            row.eval_metrics = eval_metrics
        if error is not None:
            row.error_message = error
        await db.commit()

validate_training_config

validate_training_config(
    training_config: Optional[dict],
) -> None

Reject out-of-range hyperparameters before a job is enqueued (A4.12).

Source code in adapta/services/training.py

def validate_training_config(training_config: Optional[dict]) -> None:
    """Reject out-of-range hyperparameters before a job is enqueued (A4.12)."""
    if not training_config:
        return
    for key, (lo, hi) in _HYPERPARAM_BOUNDS.items():
        if key not in training_config or training_config[key] is None:
            continue
        val = training_config[key]
        if not isinstance(val, (int, float)) or isinstance(val, bool):
            raise InvalidRequest(message=f"training_config.{key} must be a number")
        if not (lo <= val <= hi):
            raise InvalidRequest(
                message=f"training_config.{key}={val} is out of range — must be between {lo} and {hi}."
            )

check_min_training_samples

check_min_training_samples(
    num_samples: Optional[int],
) -> None

Reject a too-small dataset before a job is created (A4.6).

Below settings.min_training_samples the held-out eval split collapses (e.g. 1 row → 0 held out → the gate scores the training rows), so the eval gate would only measure memorization. Raising here keeps that signal honest.

Source code in adapta/services/training.py

def check_min_training_samples(num_samples: Optional[int]) -> None:
    """Reject a too-small dataset before a job is created (A4.6).

    Below ``settings.min_training_samples`` the held-out eval split collapses
    (e.g. 1 row → 0 held out → the gate scores the training rows), so the eval
    gate would only measure memorization. Raising here keeps that signal honest.
    """
    n = num_samples or 0
    if n < settings.min_training_samples:
        raise InvalidRequest(
            message=(
                f"Dataset has {n} sample(s); at least {settings.min_training_samples} "
                "are required to train. A smaller set can't be split into a held-out "
                "eval set, so the gate would only measure memorization."
            )
        )

recover_orphaned_jobs async

recover_orphaned_jobs(
    max_attempts: int = 1,
) -> tuple[int, int]

Reconcile jobs a crashed worker left behind (A4.2). Returns (requeued, failed).

A hard crash (OOM-kill, power loss, segfault) pops a job off the Redis queue but never drives it to a terminal state — so it sits running in Postgres forever with no live worker. A failed enqueue (or a wiped Redis) can likewise leave a job queued in Postgres but absent from the queue. On worker startup we find both and either requeue them (reconstructing the payload from Postgres, so it survives a wiped Redis) or fail them once they've burned their retries.

Single-worker assumption: at startup no other worker is live, so any running job is orphaned. Run this before the dequeue loop.

Source code in adapta/services/training.py

async def recover_orphaned_jobs(max_attempts: int = 1) -> tuple[int, int]:
    """Reconcile jobs a crashed worker left behind (A4.2). Returns (requeued, failed).

    A hard crash (OOM-kill, power loss, segfault) pops a job off the Redis queue
    but never drives it to a terminal state — so it sits ``running`` in Postgres
    forever with no live worker. A failed enqueue (or a wiped Redis) can likewise
    leave a job ``queued`` in Postgres but absent from the queue. On worker startup
    we find both and either requeue them (reconstructing the payload from Postgres,
    so it survives a wiped Redis) or fail them once they've burned their retries.

    Single-worker assumption: at startup no other worker is live, so any ``running``
    job is orphaned. Run this before the dequeue loop.
    """
    from datetime import datetime, timezone

    from sqlalchemy import select

    from adapta.db.models import JobStatus as _JobStatus
    from adapta.db.models import Project, TrainingJob
    from adapta.db.session import AsyncSessionLocal

    queue = get_job_queue()
    queued_ids = set(await queue.queued_job_ids())
    requeued = failed = 0

    async with AsyncSessionLocal() as db:
        rows = (
            (
                await db.execute(
                    select(TrainingJob).where(
                        TrainingJob.status.in_([_JobStatus.running, _JobStatus.queued])
                    )
                )
            )
            .scalars()
            .all()
        )

        for row in rows:
            # A `queued` job that's still in the Redis queue is legitimately waiting.
            if row.status == _JobStatus.queued and row.id in queued_ids:
                continue

            is_crash = row.status == _JobStatus.running
            if is_crash:
                # Only a real crash (was `running`) consumes a retry; a lost-enqueue
                # `queued` job never actually ran.
                row.attempts = (row.attempts or 0) + 1

            dataset = (
                await db.execute(select(Dataset).where(Dataset.id == row.dataset_id))
            ).scalar_one_or_none()
            project = (
                await db.execute(select(Project).where(Project.id == row.project_id))
            ).scalar_one_or_none()

            give_up = (
                (is_crash and row.attempts > max_attempts) or dataset is None or project is None
            )
            if give_up:
                reason = (
                    "dataset or project no longer exists"
                    if dataset is None or project is None
                    else f"worker crashed during training and exhausted retries ({max_attempts})"
                )
                row.status = _JobStatus.failed
                row.error_message = f"Job could not be recovered — {reason}."
                row.finished_at = datetime.now(timezone.utc)
                await db.commit()
                await queue.update_status(row.id, status="failed", error=row.error_message)
                failed += 1
                logger.warning("Orphaned job %s failed during recovery — %s", row.id, reason)
                continue

            assert dataset is not None and project is not None  # give_up covered None
            row.status = _JobStatus.queued
            row.progress = 0.0
            await db.commit()

            payload = {
                "job_id": row.id,
                "project_id": row.project_id,
                "dataset_id": row.dataset_id,
                "dataset_path": dataset.storage_path,
                "base_model": project.base_model,
                "training_config": json.loads(row.training_config) if row.training_config else {},
            }
            await queue.enqueue(row.id, payload)
            requeued += 1
            logger.warning(
                "Requeued orphaned job %s (was %s, attempt %d)",
                row.id,
                "running" if is_crash else "queued-lost",
                row.attempts,
            )

    if requeued or failed:
        logger.info("Crash recovery: %d job(s) requeued, %d failed", requeued, failed)
    return requeued, failed

enqueue_training_job async

enqueue_training_job(
    db: AsyncSession,
    project_id: str,
    dataset_id: str,
    base_model: str,
    training_config: Optional[dict] = None,
    method: str = "sft",
) -> TrainingJob

Create a TrainingJob record and push it to the Redis queue.

Source code in adapta/services/training.py

async def enqueue_training_job(
    db: AsyncSession,
    project_id: str,
    dataset_id: str,
    base_model: str,
    training_config: Optional[dict] = None,
    method: str = "sft",
) -> TrainingJob:
    """Create a TrainingJob record and push it to the Redis queue."""
    from sqlalchemy import select

    result = await db.execute(
        select(Dataset).where(Dataset.id == dataset_id, Dataset.project_id == project_id)
    )
    dataset: Optional[Dataset] = result.scalar_one_or_none()
    if not dataset:
        raise NotFound(message="Dataset not found")
    if dataset.status != DatasetStatus.valid:
        raise InvalidRequest(message="Dataset is not valid — cannot start training")
    # §V3: the dataset's modality must match the base model's. A vision dataset
    # on a text base would crash mid-train (no image inputs); a text dataset on
    # a vision base would waste the vision tower — reject both with the fix.
    from adapta.core.model_catalog import resolve as resolve_catalog

    entry = resolve_catalog(base_model)
    model_modality = entry.modality if entry else "text"
    if dataset.modality != model_modality:
        if dataset.modality == "vision":
            raise InvalidRequest(
                message=(
                    f"This dataset contains images but the project's base model "
                    f"{base_model!r} is text-only — create the project with a vision "
                    "base model (e.g. qwen2.5-vl-3b-instruct) to train on images."
                )
            )
        raise InvalidRequest(
            message=(
                f"The project's base model {base_model!r} is a vision model but this "
                "dataset has no images — upload an image bundle, or use a text base model."
            )
        )
    check_min_training_samples(dataset.num_samples)
    validate_training_config(training_config)

    # DPO is text-only: vision LoRA DPO is not supported (preference pairs with
    # image inputs require a separate evaluation protocol we don't implement yet).
    if method == "dpo" and dataset.modality != "text":
        raise InvalidRequest(
            message="DPO training is only supported for text datasets — use method=sft for vision."
        )

    job = TrainingJob(
        project_id=project_id,
        dataset_id=dataset_id,
        status=JobStatus.queued,
        training_config=json.dumps(training_config or {}),
    )
    db.add(job)
    await db.flush()

    payload = {
        "job_id": job.id,
        "project_id": project_id,
        "dataset_id": dataset_id,
        "dataset_path": dataset.storage_path,
        "base_model": base_model,
        "training_config": training_config or {},
        "method": method,
    }

    queue = get_job_queue()
    try:
        await queue.enqueue(job.id, payload)
    except RuntimeError as exc:
        raise InvalidRequest(
            message="Job queue is not available — training cannot be started right now.",
            internal_detail=str(exc),
        ) from exc
    logger.info("Training job %s enqueued for project %s", job.id, project_id)
    return job

Adapters (registry + eval gate)

Adapter registry and eval gate.

An adapter must pass the eval gate before it can back a serving endpoint. The registry is a JSON file on disk (not in the DB) so the worker can write it without opening a DB connection during the training run.

AdapterRegistry

Filesystem-based adapter registry (JSON file in the adapters directory).

Each entry: adapter_id, project_id, job_id, path, eval_score, base_model. Writes are not concurrent-safe — only one worker process ever writes.

Source code in adapta/services/adapters.py

class AdapterRegistry:
    """Filesystem-based adapter registry (JSON file in the adapters directory).

    Each entry: adapter_id, project_id, job_id, path, eval_score, base_model.
    Writes are not concurrent-safe — only one worker process ever writes.
    """

    def __init__(self, adapters_dir: Path):
        self._dir = adapters_dir
        self._dir.mkdir(parents=True, exist_ok=True)
        self._registry_path = self._dir / REGISTRY_FILE

    def _load(self) -> dict:
        if not self._registry_path.exists():
            return {}
        return json.loads(self._registry_path.read_text())  # type: ignore[no-any-return]

    def _save(self, data: dict) -> None:
        self._registry_path.write_text(json.dumps(data, indent=2))

    def register(
        self,
        adapter_id: str,
        project_id: str,
        job_id: str,
        adapter_path: str,
        eval_score: float,
        base_model: str,
        adapter_gguf_path: Optional[str] = None,
        base_score: Optional[float] = None,
        score_delta: Optional[float] = None,
    ) -> dict:
        """
        Register an adapter after it passes the eval gate.
        Raises EvalGateFailed if it passes neither the absolute nor the improvement
        path (see ``passes_eval_gate``).

        ``adapter_path`` is the PEFT directory (safetensors); ``adapter_gguf_path``
        is the converted GGUF LoRA the llama-cpp serving runtime loads.
        ``base_score``/``score_delta`` (adapter-vs-base on the held-out split) enable
        the improvement pass path.
        """
        from adapta.training.models import passes_eval_gate

        if not passes_eval_gate(eval_score, base_score, score_delta):
            threshold = settings.eval_score_threshold
            raise EvalGateFailed(
                message=(
                    f"Adapter eval score {eval_score:.3f} did not pass the gate "
                    f"(needs score ≥ {threshold:.3f}, or a clear improvement over the "
                    "base model). Adapter not registered."
                ),
                internal_detail=(
                    f"job_id={job_id}, score={eval_score}, base_score={base_score}, "
                    f"score_delta={score_delta}, threshold={threshold}"
                ),
            )

        entry = {
            "adapter_id": adapter_id,
            "project_id": project_id,
            "job_id": job_id,
            "path": adapter_path,
            "adapter_gguf_path": adapter_gguf_path,
            "eval_score": eval_score,
            "base_score": base_score,
            "score_delta": score_delta,
            "base_model": base_model,
        }
        data = self._load()
        data[adapter_id] = entry
        self._save(data)
        logger.info("Adapter %s registered (score=%.3f)", adapter_id, eval_score)
        return entry

    def get(self, adapter_id: str) -> dict:
        data = self._load()
        if adapter_id not in data:
            raise NotFound(message=f"Adapter not found: {adapter_id}")
        return data[adapter_id]  # type: ignore[no-any-return]

    def list_for_project(self, project_id: str) -> list[dict]:
        return [e for e in self._load().values() if e["project_id"] == project_id]

    def get_adapter_path(self, adapter_id: str) -> Path:
        entry = self.get(adapter_id)
        return Path(entry["path"])

register

register(
    adapter_id: str,
    project_id: str,
    job_id: str,
    adapter_path: str,
    eval_score: float,
    base_model: str,
    adapter_gguf_path: Optional[str] = None,
    base_score: Optional[float] = None,
    score_delta: Optional[float] = None,
) -> dict

Register an adapter after it passes the eval gate. Raises EvalGateFailed if it passes neither the absolute nor the improvement path (see passes_eval_gate).

adapter_path is the PEFT directory (safetensors); adapter_gguf_path is the converted GGUF LoRA the llama-cpp serving runtime loads. base_score/score_delta (adapter-vs-base on the held-out split) enable the improvement pass path.

Source code in adapta/services/adapters.py

def register(
    self,
    adapter_id: str,
    project_id: str,
    job_id: str,
    adapter_path: str,
    eval_score: float,
    base_model: str,
    adapter_gguf_path: Optional[str] = None,
    base_score: Optional[float] = None,
    score_delta: Optional[float] = None,
) -> dict:
    """
    Register an adapter after it passes the eval gate.
    Raises EvalGateFailed if it passes neither the absolute nor the improvement
    path (see ``passes_eval_gate``).

    ``adapter_path`` is the PEFT directory (safetensors); ``adapter_gguf_path``
    is the converted GGUF LoRA the llama-cpp serving runtime loads.
    ``base_score``/``score_delta`` (adapter-vs-base on the held-out split) enable
    the improvement pass path.
    """
    from adapta.training.models import passes_eval_gate

    if not passes_eval_gate(eval_score, base_score, score_delta):
        threshold = settings.eval_score_threshold
        raise EvalGateFailed(
            message=(
                f"Adapter eval score {eval_score:.3f} did not pass the gate "
                f"(needs score ≥ {threshold:.3f}, or a clear improvement over the "
                "base model). Adapter not registered."
            ),
            internal_detail=(
                f"job_id={job_id}, score={eval_score}, base_score={base_score}, "
                f"score_delta={score_delta}, threshold={threshold}"
            ),
        )

    entry = {
        "adapter_id": adapter_id,
        "project_id": project_id,
        "job_id": job_id,
        "path": adapter_path,
        "adapter_gguf_path": adapter_gguf_path,
        "eval_score": eval_score,
        "base_score": base_score,
        "score_delta": score_delta,
        "base_model": base_model,
    }
    data = self._load()
    data[adapter_id] = entry
    self._save(data)
    logger.info("Adapter %s registered (score=%.3f)", adapter_id, eval_score)
    return entry

---

Training core — the eval gate

The gate's score math and the dual (absolute-or-improvement) gate live here. See the eval gate explained.

Training data models: config, eval metrics, and the eval gate.

TrainingConfig is the operator-facing hyperparameter surface (serialized into the job payload). EvaluationMetrics / EvaluationResult carry the held-out eval signal. score_from_loss / passes_eval_gate are the gate's two-path decision (absolute floor OR clear improvement over base).

TrainingConfig dataclass

Operator-facing LoRA training hyperparameters.

All fields have sane defaults. Clients pass only the fields they want to override (exclude_unset in the router); the rest fill from these defaults.

Source code in adapta/training/models.py

@dataclass
class TrainingConfig:
    """Operator-facing LoRA training hyperparameters.

    All fields have sane defaults. Clients pass only the fields they want to
    override (``exclude_unset`` in the router); the rest fill from these defaults.
    """

    # LoRA parameters
    lora_r: int = 16  # Rank
    lora_alpha: int = 32  # Scaling factor
    lora_dropout: float = 0.05
    target_modules: List[str] = field(
        default_factory=lambda: ["q_proj", "v_proj", "k_proj", "o_proj"]
    )

    # Training parameters
    num_epochs: int = 3
    batch_size: int = 4
    learning_rate: float = 2e-4
    warmup_steps: int = 100
    gradient_accumulation_steps: int = 4
    max_seq_length: int = 2048

    # Pinned seed — recorded into training provenance so a run is reproducible.
    seed: int = 42

    # Optimizer
    optimizer: str = "adamw_torch"
    weight_decay: float = 0.01
    max_grad_norm: float = 1.0

    # Advanced options
    use_qlora: bool = True  # 4-bit quantization for memory efficiency
    gradient_checkpointing: bool = True
    logging_steps: int = 10
    eval_steps: int = 100
    save_steps: int = 500

    # Training method (D4): "sft" (default) or "dpo" (Direct Preference Optimisation).
    # Controls which TRL trainer is used and how the dataset rows are formatted.
    method: str = "sft"
    # DPO-specific: KL-penalty coefficient β. Higher values keep the fine-tuned
    # distribution close to the reference (base) model.
    dpo_beta: float = 0.1

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "TrainingConfig":
        return cls(**data)

EvaluationMetrics dataclass

Per-adapter evaluation metrics from the held-out split.

Source code in adapta/training/models.py

@dataclass
class EvaluationMetrics:
    """Per-adapter evaluation metrics from the held-out split."""

    # Core metrics — RESPONSE-ONLY cross-entropy on the HELD-OUT split (prompt
    # tokens masked). This is the signal the eval gate scores; see score_from_loss.
    loss: float
    perplexity: float

    # Accuracy metrics
    accuracy: Optional[float] = None
    exact_match: Optional[float] = None

    # Token-level metrics
    token_accuracy: Optional[float] = None
    bleu_score: Optional[float] = None

    # Quality metrics
    coherence_score: Optional[float] = None
    fluency_score: Optional[float] = None

    # Base-vs-adapter comparison on the SAME held-out split (relative signal).
    # base_loss/base_perplexity are the un-adapted base model's response-only loss;
    # loss_improvement = base_loss - loss (positive ⇒ the fine-tune helped).
    base_loss: Optional[float] = None
    base_perplexity: Optional[float] = None
    loss_improvement: Optional[float] = None

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

EvaluationResult dataclass

Complete evaluation result for one adapter run.

Source code in adapta/training/models.py

@dataclass
class EvaluationResult:
    """Complete evaluation result for one adapter run."""

    eval_id: str
    job_id: str
    agent_id: str
    adapter_name: str
    adapter_path: str

    # Dataset info
    dataset_path: str
    num_examples: int

    # Metrics
    metrics: EvaluationMetrics

    # Overall eval score in [0, 1] (higher = better), derived from the RESPONSE-ONLY
    # held-out loss via score_from_loss(). This is the value the eval gate compares to
    # the threshold (the absolute floor).
    score: float = 0.0

    # The base (un-adapted) model's score on the SAME held-out split, and the delta.
    # score_delta = score - base_score; positive ⇒ the fine-tune improved over base.
    # Reported for operator insight; the gate still uses the absolute `score`.
    base_score: Optional[float] = None
    score_delta: Optional[float] = None

    # Whether the eval ran on a held-out split (vs. fell back to the full dataset for
    # a degenerate single-row dataset). Lets the gate's meaning be read honestly.
    held_out: bool = True

    # Sample predictions (for debugging/inspection)
    sample_predictions: List[Dict[str, str]] = field(default_factory=list)

    # Metadata
    created_at: float = field(default_factory=time.time)
    duration_seconds: float = 0.0

    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary"""
        return {
            "eval_id": self.eval_id,
            "job_id": self.job_id,
            "agent_id": self.agent_id,
            "adapter_name": self.adapter_name,
            "adapter_path": self.adapter_path,
            "dataset_path": self.dataset_path,
            "num_examples": self.num_examples,
            "metrics": self.metrics.to_dict(),
            "score": self.score,
            "base_score": self.base_score,
            "score_delta": self.score_delta,
            "held_out": self.held_out,
            "sample_predictions": self.sample_predictions,
            "created_at": self.created_at,
            "duration_seconds": self.duration_seconds,
        }

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "EvaluationResult":
        data = data.copy()
        data["metrics"] = EvaluationMetrics(**data["metrics"])
        return cls(**data)

to_dict

to_dict() -> Dict[str, Any]

Convert to dictionary

Source code in adapta/training/models.py

def to_dict(self) -> Dict[str, Any]:
    """Convert to dictionary"""
    return {
        "eval_id": self.eval_id,
        "job_id": self.job_id,
        "agent_id": self.agent_id,
        "adapter_name": self.adapter_name,
        "adapter_path": self.adapter_path,
        "dataset_path": self.dataset_path,
        "num_examples": self.num_examples,
        "metrics": self.metrics.to_dict(),
        "score": self.score,
        "base_score": self.base_score,
        "score_delta": self.score_delta,
        "held_out": self.held_out,
        "sample_predictions": self.sample_predictions,
        "created_at": self.created_at,
        "duration_seconds": self.duration_seconds,
    }

score_from_loss

score_from_loss(avg_loss: float) -> float

Map an average cross-entropy loss to an eval score in [0, 1] (higher = better).

score = 1 / perplexity = exp(-avg_loss). A perfect next-token predictor (loss → 0) scores 1.0; the score decays as loss grows. This is the intrinsic language-model signal the evaluator can compute without a task-specific metric, and it is what the eval gate (eval_score_threshold, default 0.6) tests: 0.6 ⇔ perplexity ≤ ~1.67. Clamped to [0, 1] to be safe against tiny negatives.

METRIC SEMANTICS (the gate's meaning — see also specs/schemas/training_dataset.schema.json): The loss this score is built from is the response-only cross-entropy on a held-out split (rows the model never trained on), with prompt tokens masked out. So score measures how well the fine-tune generalizes at producing the target responses, not how well it memorized the training rows. The evaluator also scores the base model on the same held-out split and reports the adapter-minus-base delta (score_delta) so an operator can see whether the fine-tune actually helped; the 0.6 gate itself remains an absolute floor on the adapter's response-only score.

Source code in adapta/training/models.py

def score_from_loss(avg_loss: float) -> float:
    """Map an average cross-entropy loss to an eval score in [0, 1] (higher = better).

    ``score = 1 / perplexity = exp(-avg_loss)``. A perfect next-token predictor
    (loss → 0) scores 1.0; the score decays as loss grows. This is the intrinsic
    language-model signal the evaluator can compute without a task-specific metric,
    and it is what the eval **gate** (``eval_score_threshold``, default 0.6) tests:
    0.6 ⇔ perplexity ≤ ~1.67. Clamped to [0, 1] to be safe against tiny negatives.

    METRIC SEMANTICS (the gate's meaning — see also specs/schemas/training_dataset.schema.json):
    The loss this score is built from is the **response-only** cross-entropy on a
    **held-out** split (rows the model never trained on), with prompt tokens masked
    out. So ``score`` measures how well the fine-tune *generalizes* at producing the
    target responses, not how well it memorized the training rows. The evaluator also
    scores the **base** model on the same held-out split and reports the
    adapter-minus-base delta (``score_delta``) so an operator can see whether the
    fine-tune actually helped; the 0.6 gate itself remains an absolute floor on the
    adapter's response-only score.
    """
    if not math.isfinite(avg_loss):
        return 0.0
    return max(0.0, min(1.0, math.exp(-avg_loss)))

passes_eval_gate

passes_eval_gate(
    score: float,
    base_score: Optional[float] = None,
    score_delta: Optional[float] = None,
    *,
    threshold: Optional[float] = None,
    min_improvement: Optional[float] = None,
    min_floor: Optional[float] = None
) -> bool

The eval gate: may this adapter back an endpoint?

Two ways to pass (the moat verifies the fine-tune is good, not just that it hit one strict number):

1. Absolute — score >= threshold (a strong adapter), the original gate.
2. Improvement — the adapter clears a low sanity floor AND beats the base model on the same held-out split by min_improvement. score here is exp(-held_out_response_perplexity), which a small base model can't push to 0.6 even on an ideal task; an adapter that reliably out-scores its base by a clear margin has demonstrably learned the target behavior.

A non-improving or garbage adapter passes neither and stays blocked.

Source code in adapta/training/models.py

def passes_eval_gate(
    score: float,
    base_score: Optional[float] = None,
    score_delta: Optional[float] = None,
    *,
    threshold: Optional[float] = None,
    min_improvement: Optional[float] = None,
    min_floor: Optional[float] = None,
) -> bool:
    """The eval gate: may this adapter back an endpoint?

    Two ways to pass (the moat verifies the fine-tune is *good*, not just that it
    hit one strict number):

    1. **Absolute** — ``score >= threshold`` (a strong adapter), the original gate.
    2. **Improvement** — the adapter clears a low sanity floor AND beats the base
       model on the same held-out split by ``min_improvement``. ``score`` here is
       ``exp(-held_out_response_perplexity)``, which a small base model can't push
       to 0.6 even on an ideal task; an adapter that reliably out-scores its base by
       a clear margin has demonstrably learned the target behavior.

    A non-improving or garbage adapter passes neither and stays blocked.
    """
    from adapta.config import settings

    threshold = settings.eval_score_threshold if threshold is None else threshold
    min_improvement = settings.eval_min_improvement if min_improvement is None else min_improvement
    min_floor = settings.eval_min_floor if min_floor is None else min_floor

    if score >= threshold:
        return True
    if base_score is not None and score_delta is not None:
        return score >= min_floor and score_delta >= min_improvement
    return False

split_holdout

split_holdout(
    n: int,
    holdout_fraction: float = 0.2,
    min_holdout: int = 1,
) -> int

Return the number of trailing rows to hold out for evaluation.

The eval gate must measure generalization, so the model is scored on rows it never trained on. We hold out the last holdout_fraction of the dataset (deterministic, no shuffling needed — order is the operator's) and train on the rest. Guarantees at least min_holdout eval row and at least one training row whenever n >= 2; for a degenerate n == 1 dataset there is nothing to hold out, so it returns 0 (the caller falls back to evaluating on the single row and the absolute floor still applies).

Source code in adapta/training/models.py

def split_holdout(n: int, holdout_fraction: float = 0.2, min_holdout: int = 1) -> int:
    """Return the number of trailing rows to hold out for evaluation.

    The eval gate must measure **generalization**, so the model is scored on rows it
    never trained on. We hold out the **last** ``holdout_fraction`` of the dataset
    (deterministic, no shuffling needed — order is the operator's) and train on the
    rest. Guarantees at least ``min_holdout`` eval row and at least one training row
    whenever ``n >= 2``; for a degenerate ``n == 1`` dataset there is nothing to hold
    out, so it returns 0 (the caller falls back to evaluating on the single row and
    the absolute floor still applies).
    """
    if n <= 1:
        return 0
    k = int(round(n * holdout_fraction))
    k = max(min_holdout, k)
    k = min(k, n - 1)  # always leave at least one training row
    return k

---

Core — engine wrappers

Inference engine

Wraps llama-cpp. The per-model lock, bounded executor, and streaming bridge live here. Do not rewrite the engine internals — wrap and call.

llama-cpp-python inference engine wrapper.

Wraps the Llama object for async, serialized, timeout-bounded inference. Do NOT rewrite the llama-cpp internals here — this module calls the library; it does not replace it (hard constraint #1 in CLAUDE.md).

Chat format: all models currently use the Qwen2.5 / ChatML template (<|im_start|>role\n...<|im_end|>), hardcoded in _format_chat_prompt. Adding a model with a different chat template (Llama-3, Mistral, etc.) would require extending that function.

Message dataclass

Single chat turn.

role must be one of the OpenAI roles (user, assistant, system) — it is embedded literally into the ChatML <|im_start|>{role} token. An unrecognized role produces malformed output silently.

Source code in adapta/core/inference.py

@dataclass
class Message:
    """Single chat turn.

    ``role`` must be one of the OpenAI roles (``user``, ``assistant``,
    ``system``) — it is embedded literally into the ChatML ``<|im_start|>{role}``
    token.  An unrecognized role produces malformed output silently.
    """

    role: str
    content: str

InferenceRequest dataclass

Parameters for one inference call.

Messages are formatted via _format_chat_prompt using the Qwen2.5 ChatML template before being passed to llama-cpp.

adapter_path is the GGUF LoRA path produced by the fine-tune pipeline (not the PEFT directory). None means base/RAG serving.

Source code in adapta/core/inference.py

@dataclass
class InferenceRequest:
    """Parameters for one inference call.

    Messages are formatted via ``_format_chat_prompt`` using the Qwen2.5
    ChatML template before being passed to llama-cpp.

    ``adapter_path`` is the GGUF LoRA path produced by the fine-tune
    pipeline (not the PEFT directory).  None means base/RAG serving.
    """

    messages: List[Message]
    model_name: str
    temperature: float = settings.temperature
    top_p: float = settings.top_p
    top_k: int = settings.top_k
    max_tokens: int = settings.max_tokens
    stream: bool = True
    stop: Optional[List[str]] = None
    system_prompt: Optional[str] = None
    adapter_path: Optional[str] = None

InferenceResponse dataclass

Result of one inference call.

prompt_tokens / completion_tokens come from llama-cpp's usage dict and are accurate for text-only calls. Vision calls add image-patch tokens via a separate estimate in chat.py — the counts here undercount image-heavy traffic.

Source code in adapta/core/inference.py

@dataclass
class InferenceResponse:
    """Result of one inference call.

    ``prompt_tokens`` / ``completion_tokens`` come from llama-cpp's ``usage``
    dict and are accurate for text-only calls.  Vision calls add image-patch
    tokens via a separate estimate in ``chat.py`` — the counts here undercount
    image-heavy traffic.
    """

    content: str
    model: str
    finish_reason: str = "stop"
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0

InferenceEngine

Async wrapper around llama-cpp's Llama object.

Key invariants: - All public methods are async; blocking C calls run on _inference_executor. - A Llama object is NOT safe for concurrent calls — callers must pass the per-model serialization lock from ModelManager.get_inference_lock. - The lock is held until the underlying C call completes, even on timeout, because a llama-cpp call cannot be cancelled mid-flight.

Source code in adapta/core/inference.py

class InferenceEngine:
    """Async wrapper around llama-cpp's ``Llama`` object.

    Key invariants:
    - All public methods are async; blocking C calls run on ``_inference_executor``.
    - A ``Llama`` object is NOT safe for concurrent calls — callers must pass the
      per-model serialization lock from ``ModelManager.get_inference_lock``.
    - The lock is held until the underlying C call completes, even on timeout,
      because a llama-cpp call cannot be cancelled mid-flight.
    """

    def __init__(self):
        pass

    def _format_chat_prompt(self, request: InferenceRequest) -> str:
        """Format messages into a Qwen2.5 / ChatML prompt string.

        Hardcoded to the ChatML template (``<|im_start|>role\\n...<|im_end|>``).
        All catalog models currently use this format.  A model with a different
        chat template would need an additional branch here.
        """
        prompt_parts = []
        if request.system_prompt:
            prompt_parts.append(f"<|im_start|>system\n{request.system_prompt}<|im_end|>")
        for msg in request.messages:
            prompt_parts.append(f"<|im_start|>{msg.role}\n{msg.content}<|im_end|>")
        prompt_parts.append("<|im_start|>assistant\n")
        return "\n".join(prompt_parts)

    async def generate(
        self,
        model: Llama,
        request: InferenceRequest,
        *,
        lock: Optional[asyncio.Lock] = None,
    ) -> InferenceResponse:
        """Generate a complete response.

        `lock` (from ``model_manager.get_inference_lock``) serializes calls on
        this Llama instance (A4.1). It is held until the underlying C call truly
        returns — even on timeout — because a llama-cpp call cannot be cancelled
        and must never run concurrently with the next request on the same model.
        """
        prompt = self._format_chat_prompt(request)
        stop_tokens = request.stop or ["<|im_end|>", "<|endoftext|>"]
        loop = asyncio.get_event_loop()

        def _call():
            return model(
                prompt,
                max_tokens=request.max_tokens,
                temperature=request.temperature,
                top_p=request.top_p,
                top_k=request.top_k,
                stop=stop_tokens,
                echo=False,
            )

        try:
            result = await self._run_locked(loop, _call, lock)
        except Timeout:
            raise
        except Exception as e:
            logger.error(f"Inference error: {e}")
            raise

        choice = result["choices"][0]
        content = choice["text"].strip()
        finish_reason = choice["finish_reason"]
        usage = result.get("usage", {})
        prompt_tokens = usage.get("prompt_tokens", 0)
        completion_tokens = usage.get("completion_tokens", 0)

        return InferenceResponse(
            content=content,
            model=request.model_name,
            finish_reason=finish_reason,
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            total_tokens=prompt_tokens + completion_tokens,
        )

    async def generate_chat(
        self,
        model: Llama,
        messages: List[dict],
        *,
        model_name: str,
        temperature: float,
        top_p: float,
        max_tokens: int,
        lock: Optional[asyncio.Lock] = None,
    ) -> InferenceResponse:
        """Generate via the model's chat handler (``create_chat_completion``).

        The vision path (§V4): a multimodal model is loaded with a chat handler
        that routes ``image_url`` content-parts through the vision projector —
        something the text path's manual prompt formatting cannot express. Same
        serialization-lock and timeout semantics as ``generate`` (A4.1).
        """
        loop = asyncio.get_event_loop()

        def _call():
            return model.create_chat_completion(
                messages=messages,
                max_tokens=max_tokens,
                temperature=temperature,
                top_p=top_p,
            )

        try:
            result = await self._run_locked(loop, _call, lock)
        except Timeout:
            raise
        except Exception as e:
            logger.error(f"Chat-handler inference error: {e}")
            raise

        choice = result["choices"][0]
        content = (choice["message"].get("content") or "").strip()
        usage = result.get("usage", {})
        prompt_tokens = usage.get("prompt_tokens", 0)
        completion_tokens = usage.get("completion_tokens", 0)
        return InferenceResponse(
            content=content,
            model=model_name,
            finish_reason=choice.get("finish_reason") or "stop",
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            total_tokens=prompt_tokens + completion_tokens,
        )

    async def _run_locked(self, loop, fn, lock: Optional[asyncio.Lock]):
        """Run a blocking llama-cpp call serialized by `lock` and bounded by the
        inference timeout (A4.1).

        On timeout the call cannot be interrupted (Python can't kill the thread),
        so we never abandon it while another request might start: the model lock
        is released only when the thread actually finishes (via the done-callback),
        and `shield` keeps the future alive past the client-facing 504."""
        timeout = settings.inference_timeout_seconds
        if lock is None:
            fut = loop.run_in_executor(_inference_executor, fn)
            try:
                return await asyncio.wait_for(asyncio.shield(fut), timeout=timeout)
            except asyncio.TimeoutError as exc:
                raise Timeout(message="Inference timed out", internal_detail=str(exc)) from exc

        await lock.acquire()
        fut = loop.run_in_executor(_inference_executor, fn)
        # Release the model lock only when the C call truly returns — not when the
        # client times out — so the next same-model request can't race it.
        fut.add_done_callback(lambda _f: lock.release())
        try:
            return await asyncio.wait_for(asyncio.shield(fut), timeout=timeout)
        except asyncio.TimeoutError as exc:
            raise Timeout(message="Inference timed out", internal_detail=str(exc)) from exc

    async def generate_stream(
        self,
        model: Llama,
        request: InferenceRequest,
        *,
        lock: Optional[asyncio.Lock] = None,
    ) -> AsyncIterator[str]:
        """Generate a streaming response.

        Two safety properties over the naive version (A4.1):
        - The model `lock` is held for the *entire* stream — every token mutates
          the same Llama context, so no other request may touch this model until
          the stream is fully drained.
        - Each token is pulled on the bounded pool (not iterated synchronously in
          the event loop), so streaming no longer blocks the whole server, and a
          wall-clock deadline is checked *between* tokens. The check is between
          tokens (never mid-call) so stopping is race-free: there is no in-flight
          C call when we close the generator and release the lock.
        """
        prompt = self._format_chat_prompt(request)
        stop_tokens = request.stop or ["<|im_end|>", "<|endoftext|>"]
        loop = asyncio.get_event_loop()

        def _create_stream():
            return model(
                prompt,
                max_tokens=request.max_tokens,
                temperature=request.temperature,
                top_p=request.top_p,
                top_k=request.top_k,
                stop=stop_tokens,
                stream=True,
                echo=False,
            )

        _SENTINEL = object()

        if lock is not None:
            await lock.acquire()
        try:
            deadline = time.monotonic() + settings.inference_timeout_seconds
            stream = await loop.run_in_executor(_inference_executor, _create_stream)

            def _next():
                try:
                    return next(stream)
                except StopIteration:
                    return _SENTINEL

            while True:
                if time.monotonic() > deadline:
                    # No `_next` is in flight here (the previous one returned), so
                    # closing + releasing the lock cannot race a live C call.
                    _close_stream(stream)
                    raise Timeout(message="Streaming inference timed out")
                chunk = await loop.run_in_executor(_inference_executor, _next)
                if chunk is _SENTINEL:
                    break
                if "choices" in chunk and len(chunk["choices"]) > 0:
                    delta = chunk["choices"][0].get("text", "")
                    if delta:
                        yield delta
        except Timeout:
            raise
        except Exception as e:
            logger.error(f"Streaming inference error: {e}")
            raise
        finally:
            if lock is not None:
                lock.release()

    def count_tokens(self, model: Llama, text: str) -> int:
        """Count tokens in ``text`` using the model's real tokenizer.

        Falls back to ``len(text) // 4`` on tokenizer error.  Callers using
        this for context budgeting should be aware the fallback may under- or
        over-count for non-ASCII text.
        """
        try:
            tokens = model.tokenize(text.encode("utf-8"))
            return len(tokens)
        except Exception as e:
            logger.warning("Token counting error: %s", e)
            return len(text) // 4

    def count_prompt_tokens(
        self, model: Llama, messages: List[Message], system_prompt: Optional[str] = None
    ) -> int:
        """Token count of the FINAL formatted prompt, via the model's real tokenizer.

        This is the exact text the model will be conditioned on, so it's the number
        to budget against ``n_ctx`` (A4.3) — not a char/4 estimate."""
        req = InferenceRequest(messages=messages, model_name="", system_prompt=system_prompt)
        return self.count_tokens(model, self._format_chat_prompt(req))

    def context_size(self, model: Llama) -> int:
        """Return the model's active context window (n_ctx).

        Falls back to ``settings.max_context_length`` on error.  If the GGUF
        was loaded with a different ``n_ctx`` than the config default, this
        returns the actual value; the fallback may not match, causing the
        context-fit guard in ``chat.py`` to use an incorrect ceiling.
        """
        try:
            return int(model.n_ctx())
        except Exception:  # pragma: no cover - defensive
            return settings.max_context_length

generate async

generate(
    model: Llama,
    request: InferenceRequest,
    *,
    lock: Optional[Lock] = None
) -> InferenceResponse

Generate a complete response.

lock (from model_manager.get_inference_lock) serializes calls on this Llama instance (A4.1). It is held until the underlying C call truly returns — even on timeout — because a llama-cpp call cannot be cancelled and must never run concurrently with the next request on the same model.

Source code in adapta/core/inference.py

async def generate(
    self,
    model: Llama,
    request: InferenceRequest,
    *,
    lock: Optional[asyncio.Lock] = None,
) -> InferenceResponse:
    """Generate a complete response.

    `lock` (from ``model_manager.get_inference_lock``) serializes calls on
    this Llama instance (A4.1). It is held until the underlying C call truly
    returns — even on timeout — because a llama-cpp call cannot be cancelled
    and must never run concurrently with the next request on the same model.
    """
    prompt = self._format_chat_prompt(request)
    stop_tokens = request.stop or ["<|im_end|>", "<|endoftext|>"]
    loop = asyncio.get_event_loop()

    def _call():
        return model(
            prompt,
            max_tokens=request.max_tokens,
            temperature=request.temperature,
            top_p=request.top_p,
            top_k=request.top_k,
            stop=stop_tokens,
            echo=False,
        )

    try:
        result = await self._run_locked(loop, _call, lock)
    except Timeout:
        raise
    except Exception as e:
        logger.error(f"Inference error: {e}")
        raise

    choice = result["choices"][0]
    content = choice["text"].strip()
    finish_reason = choice["finish_reason"]
    usage = result.get("usage", {})
    prompt_tokens = usage.get("prompt_tokens", 0)
    completion_tokens = usage.get("completion_tokens", 0)

    return InferenceResponse(
        content=content,
        model=request.model_name,
        finish_reason=finish_reason,
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        total_tokens=prompt_tokens + completion_tokens,
    )

generate_chat async

generate_chat(
    model: Llama,
    messages: List[dict],
    *,
    model_name: str,
    temperature: float,
    top_p: float,
    max_tokens: int,
    lock: Optional[Lock] = None
) -> InferenceResponse

Generate via the model's chat handler (create_chat_completion).

The vision path (§V4): a multimodal model is loaded with a chat handler that routes image_url content-parts through the vision projector — something the text path's manual prompt formatting cannot express. Same serialization-lock and timeout semantics as generate (A4.1).

Source code in adapta/core/inference.py

async def generate_chat(
    self,
    model: Llama,
    messages: List[dict],
    *,
    model_name: str,
    temperature: float,
    top_p: float,
    max_tokens: int,
    lock: Optional[asyncio.Lock] = None,
) -> InferenceResponse:
    """Generate via the model's chat handler (``create_chat_completion``).

    The vision path (§V4): a multimodal model is loaded with a chat handler
    that routes ``image_url`` content-parts through the vision projector —
    something the text path's manual prompt formatting cannot express. Same
    serialization-lock and timeout semantics as ``generate`` (A4.1).
    """
    loop = asyncio.get_event_loop()

    def _call():
        return model.create_chat_completion(
            messages=messages,
            max_tokens=max_tokens,
            temperature=temperature,
            top_p=top_p,
        )

    try:
        result = await self._run_locked(loop, _call, lock)
    except Timeout:
        raise
    except Exception as e:
        logger.error(f"Chat-handler inference error: {e}")
        raise

    choice = result["choices"][0]
    content = (choice["message"].get("content") or "").strip()
    usage = result.get("usage", {})
    prompt_tokens = usage.get("prompt_tokens", 0)
    completion_tokens = usage.get("completion_tokens", 0)
    return InferenceResponse(
        content=content,
        model=model_name,
        finish_reason=choice.get("finish_reason") or "stop",
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        total_tokens=prompt_tokens + completion_tokens,
    )

generate_stream async

generate_stream(
    model: Llama,
    request: InferenceRequest,
    *,
    lock: Optional[Lock] = None
) -> AsyncIterator[str]

Generate a streaming response.

Two safety properties over the naive version (A4.1): - The model lock is held for the entire stream — every token mutates the same Llama context, so no other request may touch this model until the stream is fully drained. - Each token is pulled on the bounded pool (not iterated synchronously in the event loop), so streaming no longer blocks the whole server, and a wall-clock deadline is checked between tokens. The check is between tokens (never mid-call) so stopping is race-free: there is no in-flight C call when we close the generator and release the lock.

Source code in adapta/core/inference.py

async def generate_stream(
    self,
    model: Llama,
    request: InferenceRequest,
    *,
    lock: Optional[asyncio.Lock] = None,
) -> AsyncIterator[str]:
    """Generate a streaming response.

    Two safety properties over the naive version (A4.1):
    - The model `lock` is held for the *entire* stream — every token mutates
      the same Llama context, so no other request may touch this model until
      the stream is fully drained.
    - Each token is pulled on the bounded pool (not iterated synchronously in
      the event loop), so streaming no longer blocks the whole server, and a
      wall-clock deadline is checked *between* tokens. The check is between
      tokens (never mid-call) so stopping is race-free: there is no in-flight
      C call when we close the generator and release the lock.
    """
    prompt = self._format_chat_prompt(request)
    stop_tokens = request.stop or ["<|im_end|>", "<|endoftext|>"]
    loop = asyncio.get_event_loop()

    def _create_stream():
        return model(
            prompt,
            max_tokens=request.max_tokens,
            temperature=request.temperature,
            top_p=request.top_p,
            top_k=request.top_k,
            stop=stop_tokens,
            stream=True,
            echo=False,
        )

    _SENTINEL = object()

    if lock is not None:
        await lock.acquire()
    try:
        deadline = time.monotonic() + settings.inference_timeout_seconds
        stream = await loop.run_in_executor(_inference_executor, _create_stream)

        def _next():
            try:
                return next(stream)
            except StopIteration:
                return _SENTINEL

        while True:
            if time.monotonic() > deadline:
                # No `_next` is in flight here (the previous one returned), so
                # closing + releasing the lock cannot race a live C call.
                _close_stream(stream)
                raise Timeout(message="Streaming inference timed out")
            chunk = await loop.run_in_executor(_inference_executor, _next)
            if chunk is _SENTINEL:
                break
            if "choices" in chunk and len(chunk["choices"]) > 0:
                delta = chunk["choices"][0].get("text", "")
                if delta:
                    yield delta
    except Timeout:
        raise
    except Exception as e:
        logger.error(f"Streaming inference error: {e}")
        raise
    finally:
        if lock is not None:
            lock.release()

count_tokens

count_tokens(model: Llama, text: str) -> int

Count tokens in text using the model's real tokenizer.

Falls back to len(text) // 4 on tokenizer error. Callers using this for context budgeting should be aware the fallback may under- or over-count for non-ASCII text.

Source code in adapta/core/inference.py

def count_tokens(self, model: Llama, text: str) -> int:
    """Count tokens in ``text`` using the model's real tokenizer.

    Falls back to ``len(text) // 4`` on tokenizer error.  Callers using
    this for context budgeting should be aware the fallback may under- or
    over-count for non-ASCII text.
    """
    try:
        tokens = model.tokenize(text.encode("utf-8"))
        return len(tokens)
    except Exception as e:
        logger.warning("Token counting error: %s", e)
        return len(text) // 4

count_prompt_tokens

count_prompt_tokens(
    model: Llama,
    messages: List[Message],
    system_prompt: Optional[str] = None,
) -> int

Token count of the FINAL formatted prompt, via the model's real tokenizer.

This is the exact text the model will be conditioned on, so it's the number to budget against n_ctx (A4.3) — not a char/4 estimate.

Source code in adapta/core/inference.py

def count_prompt_tokens(
    self, model: Llama, messages: List[Message], system_prompt: Optional[str] = None
) -> int:
    """Token count of the FINAL formatted prompt, via the model's real tokenizer.

    This is the exact text the model will be conditioned on, so it's the number
    to budget against ``n_ctx`` (A4.3) — not a char/4 estimate."""
    req = InferenceRequest(messages=messages, model_name="", system_prompt=system_prompt)
    return self.count_tokens(model, self._format_chat_prompt(req))

context_size

context_size(model: Llama) -> int

Return the model's active context window (n_ctx).

Falls back to settings.max_context_length on error. If the GGUF was loaded with a different n_ctx than the config default, this returns the actual value; the fallback may not match, causing the context-fit guard in chat.py to use an incorrect ceiling.

Source code in adapta/core/inference.py

def context_size(self, model: Llama) -> int:
    """Return the model's active context window (n_ctx).

    Falls back to ``settings.max_context_length`` on error.  If the GGUF
    was loaded with a different ``n_ctx`` than the config default, this
    returns the actual value; the fallback may not match, causing the
    context-fit guard in ``chat.py`` to use an incorrect ceiling.
    """
    try:
        return int(model.n_ctx())
    except Exception:  # pragma: no cover - defensive
        return settings.max_context_length

Model manager (cache + locks)

LRU model cache and lifecycle manager.

Maintains an OrderedDict-based LRU cache of loaded Llama objects, bounded by settings.max_loaded_models. Cache keys combine the serving (GGUF) model name and the adapter path so a fine-tune endpoint (base+LoRA) and base-only serving of the same base are cached independently.

ModelManager is instantiated once at module level as the process-level singleton; import it via from adapta.core import model_manager.

ModelType

Bases: str, Enum

Capability class of a catalog model. Used to group entries in the console.

Source code in adapta/core/model_manager.py

class ModelType(str, Enum):
    """Capability class of a catalog model.  Used to group entries in the console."""

    CHAT = "chat"
    CODE = "code"
    REASONING = "reasoning"

ModelConfig dataclass

Static configuration for one GGUF model entry.

loaded is mutable state: it is flipped to True after the model is successfully loaded into the LRU cache, and back to False on eviction.

Source code in adapta/core/model_manager.py

@dataclass
class ModelConfig:
    """Static configuration for one GGUF model entry.

    ``loaded`` is mutable state: it is flipped to True after the model is
    successfully loaded into the LRU cache, and back to False on eviction.
    """

    name: str
    model_type: ModelType
    path: Path
    context_length: int = 4096
    n_threads: int = 8
    n_gpu_layers: int = 0
    description: str = ""
    loaded: bool = False
    # Vision (§V4): path to the mmproj (vision projector) GGUF. When set, the
    # model loads with a multimodal chat handler and can take image content-parts.
    mmproj_path: Optional[Path] = None

ModelManager

LRU cache of loaded Llama objects with serialized per-model inference.

Two locks with distinct scopes: - _load_lock: guards mutations to _models / _configs (loading, eviction, cache-key bookkeeping). Held only during load, never during inference. - _infer_locks[cache_key]: serializes inference calls on one model variant (A4.1). Held for the entire duration of a generation (including streaming) because llama-cpp's Llama object is not thread-safe.

The OrderedDict ordering is most-recently-used-last so popitem(last=False) evicts the least-recently-used entry.

Source code in adapta/core/model_manager.py

class ModelManager:
    """LRU cache of loaded ``Llama`` objects with serialized per-model inference.

    Two locks with distinct scopes:
    - ``_load_lock``: guards mutations to ``_models`` / ``_configs`` (loading,
      eviction, cache-key bookkeeping).  Held only during load, never during
      inference.
    - ``_infer_locks[cache_key]``: serializes inference calls on one model
      variant (A4.1).  Held for the entire duration of a generation (including
      streaming) because llama-cpp's ``Llama`` object is not thread-safe.

    The ``OrderedDict`` ordering is most-recently-used-last so
    ``popitem(last=False)`` evicts the least-recently-used entry.
    """

    def __init__(self):
        # LRU-ordered (most-recently-used last) so the cache can be bounded (A4.8).
        self._models: "OrderedDict[str, Llama]" = OrderedDict()
        self._configs: Dict[str, ModelConfig] = {}
        self._load_lock = asyncio.Lock()
        # Per-model-variant serialization locks (A4.1). Keyed on the same (base,
        # adapter) cache key as `_models`; `_load_lock` only guards loading, not
        # the (thread-unsafe) inference call on a shared Llama instance.
        self._infer_locks: Dict[str, asyncio.Lock] = {}
        self._init_default_configs()

    def get_inference_lock(
        self, model_name: str, adapter_path: Optional[str] = None
    ) -> asyncio.Lock:
        """Return the serialization lock for a loaded model variant (A4.1).

        llama-cpp's ``Llama`` object is not safe for concurrent calls on one
        instance, so a caller MUST hold this lock for the *entire* duration of a
        generation — and across a stream's *full* consumption (each token mutates
        the same context). Keyed on the same (base, adapter) cache key as the
        model, so base/RAG and a fine-tune (base+LoRA) serialize independently."""
        cache_key = self._cache_key(self._resolve_serving_name(model_name), adapter_path)
        lock = self._infer_locks.get(cache_key)
        if lock is None:
            lock = asyncio.Lock()
            self._infer_locks[cache_key] = lock
        return lock

    def _resolve_serving_name(self, model_name: str) -> str:
        """Map an operator-facing base id (possibly a HF repo id from a fine-tune
        Project) to a GGUF catalog name. Direct serving-config names pass through.

        Resolution goes through the single base-model catalog (A3.3) so the GGUF
        served is the SAME entry the trainer/evaluator resolved the HF id from.
        Imported lazily to avoid the catalog↔model_manager import cycle (the catalog
        imports ``ModelType`` from here)."""
        if model_name in self._configs:
            return model_name
        from adapta.core.model_catalog import resolve_serving_name

        return resolve_serving_name(model_name)

    def _find_model_file(self, model_dir: Path, preferred_filename: str) -> Optional[Path]:
        """
        Find a model file in a directory.
        First tries the preferred filename, then searches for any .gguf file.
        """
        # Try preferred file first
        preferred_path = model_dir / preferred_filename
        if preferred_path.exists():
            return preferred_path

        # Search for any .gguf file in the directory
        if model_dir.exists():
            gguf_files = list(model_dir.glob("*.gguf"))
            if gguf_files:
                # Prefer Q4_K_M, then Q5_K_M, then any other
                for pattern in ["*q4_k_m.gguf", "*q5_k_m.gguf", "*.gguf"]:
                    matches = list(model_dir.glob(pattern))
                    if matches:
                        return matches[0]

        return None

    def _init_default_configs(self) -> None:
        """Populate ``_configs`` with static entries for all catalog models.

        All entries are lazy-loaded; none are pre-warmed here.  ``preload_default_models``
        optionally warms the configured default model at startup.
        """
        models_dir = settings.models_dir

        self._configs["qwen2.5-3b-instruct"] = ModelConfig(
            name="qwen2.5-3b-instruct",
            model_type=ModelType.CHAT,
            path=models_dir / "qwen2.5-3b" / "qwen2.5-3b-instruct-q4_k_m.gguf",
            context_length=32768,
            n_threads=settings.n_threads,
            n_gpu_layers=settings.n_gpu_layers,
            description="General chat and reasoning model",
        )

        self._configs["qwen2.5-coder-3b"] = ModelConfig(
            name="qwen2.5-coder-3b",
            model_type=ModelType.CODE,
            path=models_dir / "qwen2.5-coder-3b" / "qwen2.5-coder-3b-instruct-q4_k_m.gguf",
            context_length=32768,
            n_threads=settings.n_threads,
            n_gpu_layers=settings.n_gpu_layers,
            description="Code understanding and generation model",
        )

        self._configs["qwen2.5-0.5b-instruct"] = ModelConfig(
            name="qwen2.5-0.5b-instruct",
            model_type=ModelType.CHAT,
            path=models_dir / "qwen2.5-0.5b" / "qwen2.5-0.5b-instruct-q4_k_m.gguf",
            context_length=32768,
            n_threads=settings.n_threads,
            n_gpu_layers=settings.n_gpu_layers,
            description="Small instruct model (fine-tune e2e base)",
        )

        # Vision: requires mmproj so the model loads with a multimodal chat handler.
        self._configs["qwen2.5-vl-3b-instruct"] = ModelConfig(
            name="qwen2.5-vl-3b-instruct",
            model_type=ModelType.CHAT,
            path=models_dir / "qwen2.5-vl-3b" / "qwen2.5-vl-3b-instruct-q4_k_m.gguf",
            context_length=32768,
            n_threads=settings.n_threads,
            n_gpu_layers=settings.n_gpu_layers,
            description="Vision model (image+text→text) — document AI / visual QC",
            mmproj_path=models_dir / "qwen2.5-vl-3b" / "mmproj-qwen2.5-vl-3b-f16.gguf",
        )

        self._configs["qwen2.5-7b-instruct"] = ModelConfig(
            name="qwen2.5-7b-instruct",
            model_type=ModelType.REASONING,
            path=models_dir / "qwen2.5-7b" / "qwen2.5-7b-instruct-q4_k_m.gguf",
            context_length=32768,
            n_threads=settings.n_threads,
            n_gpu_layers=settings.n_gpu_layers,
            description="Large model for complex reasoning",
        )

    def _evict_lru_if_needed(self) -> None:
        """Evict least-recently-used models until there's room for one more (A4.8).

        Called under ``_load_lock`` right before inserting a NEW model. Dropping the
        dict entry only releases THIS class's reference — an in-flight request still
        holds its own reference to the Llama (chat.py keeps ``model_obj`` for the
        call), so the native memory is freed by refcounting only once no request is
        using it. The evicted endpoint transparently reloads on its next call."""
        while len(self._models) >= settings.max_loaded_models:
            old_key, _ = self._models.popitem(last=False)  # LRU = first
            self._infer_locks.pop(old_key, None)
            base = old_key.split("::lora::", 1)[0]
            if base in self._configs:
                self._configs[base].loaded = False
            logger.info(
                "Evicting LRU model %s (loaded-model cap=%d reached)",
                old_key,
                settings.max_loaded_models,
            )

    def _cache_key(self, model_name: str, adapter_path: Optional[str]) -> str:
        """Cache key for a loaded Llama. Includes the adapter so a fine-tune
        endpoint (base+LoRA) and base/RAG serving of the same base never collide
        (A3.1). Without the adapter in the key, the first-loaded variant would be
        returned for both. ``model_name`` is the resolved serving (GGUF) name."""
        return model_name if not adapter_path else f"{model_name}::lora::{adapter_path}"

    async def load_model(
        self,
        model_name: str,
        force_reload: bool = False,
        adapter_path: Optional[str] = None,
    ) -> Llama:
        """Load a model into memory, optionally applying a GGUF LoRA adapter.

        When ``adapter_path`` is given (a GGUF LoRA produced by the fine-tune
        pipeline, A3.1), the base GGUF is loaded WITH the adapter via llama-cpp's
        ``lora_path``. The cache is keyed on (base, adapter)."""
        async with self._load_lock:
            serving_name = self._resolve_serving_name(model_name)
            cache_key = self._cache_key(serving_name, adapter_path)
            if cache_key in self._models and not force_reload:
                logger.info("Model %s already loaded", cache_key)
                self._models.move_to_end(cache_key)  # mark most-recently-used (A4.8)
                return self._models[cache_key]

            if serving_name not in self._configs:
                raise ValueError(f"Unknown model: {model_name}")

            config = self._configs[serving_name]

            model_path = config.path
            if not model_path.exists():
                model_dir = model_path.parent
                preferred_filename = model_path.name
                found_path = self._find_model_file(model_dir, preferred_filename)

                if found_path:
                    logger.info(
                        f"Preferred file {model_path.name} not found, using {found_path.name} instead"
                    )
                    model_path = found_path
                    config.path = found_path  # Update config with actual path
                else:
                    raise FileNotFoundError(
                        f"Model file not found: {config.path}\n"
                        "Download the base GGUF into the models volume first "
                        "(huggingface-cli; see README §4 'Download a base model')."
                    )

            lora_path: Optional[str] = None
            if adapter_path:
                lora_file = Path(adapter_path)
                if not lora_file.exists():
                    raise FileNotFoundError(
                        f"Adapter (GGUF LoRA) not found: {adapter_path}. "
                        "The fine-tune adapter was not converted/registered for serving."
                    )
                lora_path = str(lora_file)

            logger.info(
                f"Loading model {model_name} from {model_path}"
                + (f" with LoRA adapter {lora_path}" if lora_path else "")
            )

            try:
                from adapta.core.gpu import get_model_kwargs

                gpu_kwargs = get_model_kwargs()
                if config.n_gpu_layers > 0:
                    gpu_kwargs["n_gpu_layers"] = config.n_gpu_layers
                n_ctx = gpu_kwargs.get("n_ctx", config.context_length)

                logger.info(f"GPU layers: {gpu_kwargs.get('n_gpu_layers', 0)}, Context: {n_ctx}")

                llama_kwargs = dict(
                    model_path=str(model_path),
                    n_ctx=n_ctx,
                    n_threads=config.n_threads,
                    n_gpu_layers=gpu_kwargs.get("n_gpu_layers", config.n_gpu_layers),
                    n_batch=gpu_kwargs.get("n_batch", 512),
                    f16_kv=gpu_kwargs.get("f16_kv", False),
                    use_mmap=settings.use_mmap,
                    use_mlock=settings.use_mlock,
                    verbose=False,
                )
                if lora_path:
                    llama_kwargs["lora_path"] = lora_path
                # Vision (§V4): attach the multimodal chat handler so image
                # content-parts route through the vision projector. The handler
                # is per-Llama (it owns a clip context) — never shared.
                if config.mmproj_path is not None:
                    if not config.mmproj_path.exists():
                        raise FileNotFoundError(
                            f"Vision projector (mmproj) not found: {config.mmproj_path}. "
                            "Download it next to the base GGUF."
                        )
                    if settings.n_gpu_layers == 0:
                        # A VLM (base GGUF + CLIP projector) on CPU is impractically slow for
                        # production — a single image request can take tens of seconds and tie
                        # up the model's serialization lock. Tell the operator loudly to offload.
                        logger.warning(
                            "Serving VISION model %s on CPU (ADAPTA_N_GPU_LAYERS=0): image "
                            "inference will be very slow. Set ADAPTA_N_GPU_LAYERS>0 (a GPU host) "
                            "for production vision serving.",
                            model_name,
                        )
                    from llama_cpp.llama_chat_format import Qwen25VLChatHandler

                    llama_kwargs["chat_handler"] = Qwen25VLChatHandler(
                        clip_model_path=str(config.mmproj_path), verbose=False
                    )
                loop = asyncio.get_event_loop()
                model = await loop.run_in_executor(None, lambda: Llama(**llama_kwargs))

                # force_reload replaces an existing key in-place (no net growth).
                if cache_key not in self._models:
                    self._evict_lru_if_needed()
                self._models[cache_key] = model
                self._models.move_to_end(cache_key)
                config.loaded = True
                logger.info(f"Successfully loaded model {cache_key}")
                return model

            except Exception as e:
                logger.error(f"Failed to load model {cache_key}: {e}")
                raise

    async def unload_model(self, model_name: str):
        """Unload a model from memory. Accepts a catalog name or a composite
        base+LoRA cache key (A3.1)."""
        if model_name in self._models:
            logger.info(f"Unloading model {model_name}")
            del self._models[model_name]
            self._infer_locks.pop(model_name, None)  # drop its serialization lock (A4.1)
            # The cache key may be a composite "name::lora::path"; only the base
            # catalog name has a config to flip back to unloaded.
            base = model_name.split("::lora::", 1)[0]
            if base in self._configs:
                self._configs[base].loaded = False

    def get_model(self, model_name: str) -> Optional[Llama]:
        """Return a loaded model by cache key, or None if not loaded."""
        return self._models.get(model_name)

    def list_models(self) -> list[ModelConfig]:
        """List all available models with updated paths and existence status"""
        models = []
        for config in self._configs.values():
            # Create a copy to avoid modifying the original
            model_copy = ModelConfig(
                name=config.name,
                model_type=config.model_type,
                path=config.path,
                context_length=config.context_length,
                n_threads=config.n_threads,
                n_gpu_layers=config.n_gpu_layers,
                description=config.description,
                loaded=config.loaded,
            )

            # Check if configured path exists, otherwise search for alternatives
            if not model_copy.path.exists():
                model_dir = model_copy.path.parent
                preferred_filename = model_copy.path.name
                found_path = self._find_model_file(model_dir, preferred_filename)
                if found_path:
                    model_copy.path = found_path

            models.append(model_copy)
        return models

    def get_model_config(self, model_name: str) -> Optional[ModelConfig]:
        """Get model configuration"""
        return self._configs.get(model_name)

    def get_model_by_type(self, model_type: ModelType) -> Optional[str]:
        """Get the first available model of a given type"""
        for name, config in self._configs.items():
            if config.model_type == model_type:
                # Check if file exists, or try to find an alternative
                if config.path.exists():
                    return name
                else:
                    found_path = self._find_model_file(config.path.parent, config.path.name)
                    if found_path:
                        return name
        return None

    async def ensure_model_loaded(
        self, model_name: str, adapter_path: Optional[str] = None
    ) -> Llama:
        """Return the loaded model, loading it first if absent.

        The ``cache_key not in _models`` check is not atomic with the subsequent
        ``load_model`` call.  Two concurrent callers can both see the key absent
        and both call ``load_model`` — the second completes inside ``_load_lock``
        and safely overwrites the first.  The double-load is harmless but visible
        as two consecutive log lines.
        """
        cache_key = self._cache_key(self._resolve_serving_name(model_name), adapter_path)
        if cache_key not in self._models:
            return await self.load_model(model_name, adapter_path=adapter_path)
        self._models.move_to_end(cache_key)  # mark most-recently-used (A4.8)
        return self._models[cache_key]

    def is_loaded(self, model_name: str, adapter_path: Optional[str] = None) -> bool:
        """Check if a model (optionally a specific base+LoRA variant) is loaded"""
        return self._cache_key(self._resolve_serving_name(model_name), adapter_path) in self._models

    async def preload_default_models(self):
        """Preload default models (chat model)"""
        default_model = settings.default_model
        if default_model in self._configs:
            try:
                await self.load_model(default_model)
                logger.info(f"Preloaded default model: {default_model}")
            except Exception as e:
                logger.warning(f"Could not preload default model: {e}")

get_inference_lock

get_inference_lock(
    model_name: str, adapter_path: Optional[str] = None
) -> asyncio.Lock

Return the serialization lock for a loaded model variant (A4.1).

llama-cpp's Llama object is not safe for concurrent calls on one instance, so a caller MUST hold this lock for the entire duration of a generation — and across a stream's full consumption (each token mutates the same context). Keyed on the same (base, adapter) cache key as the model, so base/RAG and a fine-tune (base+LoRA) serialize independently.

Source code in adapta/core/model_manager.py

def get_inference_lock(
    self, model_name: str, adapter_path: Optional[str] = None
) -> asyncio.Lock:
    """Return the serialization lock for a loaded model variant (A4.1).

    llama-cpp's ``Llama`` object is not safe for concurrent calls on one
    instance, so a caller MUST hold this lock for the *entire* duration of a
    generation — and across a stream's *full* consumption (each token mutates
    the same context). Keyed on the same (base, adapter) cache key as the
    model, so base/RAG and a fine-tune (base+LoRA) serialize independently."""
    cache_key = self._cache_key(self._resolve_serving_name(model_name), adapter_path)
    lock = self._infer_locks.get(cache_key)
    if lock is None:
        lock = asyncio.Lock()
        self._infer_locks[cache_key] = lock
    return lock

load_model async

load_model(
    model_name: str,
    force_reload: bool = False,
    adapter_path: Optional[str] = None,
) -> Llama

Load a model into memory, optionally applying a GGUF LoRA adapter.

When adapter_path is given (a GGUF LoRA produced by the fine-tune pipeline, A3.1), the base GGUF is loaded WITH the adapter via llama-cpp's lora_path. The cache is keyed on (base, adapter).

Source code in adapta/core/model_manager.py

async def load_model(
    self,
    model_name: str,
    force_reload: bool = False,
    adapter_path: Optional[str] = None,
) -> Llama:
    """Load a model into memory, optionally applying a GGUF LoRA adapter.

    When ``adapter_path`` is given (a GGUF LoRA produced by the fine-tune
    pipeline, A3.1), the base GGUF is loaded WITH the adapter via llama-cpp's
    ``lora_path``. The cache is keyed on (base, adapter)."""
    async with self._load_lock:
        serving_name = self._resolve_serving_name(model_name)
        cache_key = self._cache_key(serving_name, adapter_path)
        if cache_key in self._models and not force_reload:
            logger.info("Model %s already loaded", cache_key)
            self._models.move_to_end(cache_key)  # mark most-recently-used (A4.8)
            return self._models[cache_key]

        if serving_name not in self._configs:
            raise ValueError(f"Unknown model: {model_name}")

        config = self._configs[serving_name]

        model_path = config.path
        if not model_path.exists():
            model_dir = model_path.parent
            preferred_filename = model_path.name
            found_path = self._find_model_file(model_dir, preferred_filename)

            if found_path:
                logger.info(
                    f"Preferred file {model_path.name} not found, using {found_path.name} instead"
                )
                model_path = found_path
                config.path = found_path  # Update config with actual path
            else:
                raise FileNotFoundError(
                    f"Model file not found: {config.path}\n"
                    "Download the base GGUF into the models volume first "
                    "(huggingface-cli; see README §4 'Download a base model')."
                )

        lora_path: Optional[str] = None
        if adapter_path:
            lora_file = Path(adapter_path)
            if not lora_file.exists():
                raise FileNotFoundError(
                    f"Adapter (GGUF LoRA) not found: {adapter_path}. "
                    "The fine-tune adapter was not converted/registered for serving."
                )
            lora_path = str(lora_file)

        logger.info(
            f"Loading model {model_name} from {model_path}"
            + (f" with LoRA adapter {lora_path}" if lora_path else "")
        )

        try:
            from adapta.core.gpu import get_model_kwargs

            gpu_kwargs = get_model_kwargs()
            if config.n_gpu_layers > 0:
                gpu_kwargs["n_gpu_layers"] = config.n_gpu_layers
            n_ctx = gpu_kwargs.get("n_ctx", config.context_length)

            logger.info(f"GPU layers: {gpu_kwargs.get('n_gpu_layers', 0)}, Context: {n_ctx}")

            llama_kwargs = dict(
                model_path=str(model_path),
                n_ctx=n_ctx,
                n_threads=config.n_threads,
                n_gpu_layers=gpu_kwargs.get("n_gpu_layers", config.n_gpu_layers),
                n_batch=gpu_kwargs.get("n_batch", 512),
                f16_kv=gpu_kwargs.get("f16_kv", False),
                use_mmap=settings.use_mmap,
                use_mlock=settings.use_mlock,
                verbose=False,
            )
            if lora_path:
                llama_kwargs["lora_path"] = lora_path
            # Vision (§V4): attach the multimodal chat handler so image
            # content-parts route through the vision projector. The handler
            # is per-Llama (it owns a clip context) — never shared.
            if config.mmproj_path is not None:
                if not config.mmproj_path.exists():
                    raise FileNotFoundError(
                        f"Vision projector (mmproj) not found: {config.mmproj_path}. "
                        "Download it next to the base GGUF."
                    )
                if settings.n_gpu_layers == 0:
                    # A VLM (base GGUF + CLIP projector) on CPU is impractically slow for
                    # production — a single image request can take tens of seconds and tie
                    # up the model's serialization lock. Tell the operator loudly to offload.
                    logger.warning(
                        "Serving VISION model %s on CPU (ADAPTA_N_GPU_LAYERS=0): image "
                        "inference will be very slow. Set ADAPTA_N_GPU_LAYERS>0 (a GPU host) "
                        "for production vision serving.",
                        model_name,
                    )
                from llama_cpp.llama_chat_format import Qwen25VLChatHandler

                llama_kwargs["chat_handler"] = Qwen25VLChatHandler(
                    clip_model_path=str(config.mmproj_path), verbose=False
                )
            loop = asyncio.get_event_loop()
            model = await loop.run_in_executor(None, lambda: Llama(**llama_kwargs))

            # force_reload replaces an existing key in-place (no net growth).
            if cache_key not in self._models:
                self._evict_lru_if_needed()
            self._models[cache_key] = model
            self._models.move_to_end(cache_key)
            config.loaded = True
            logger.info(f"Successfully loaded model {cache_key}")
            return model

        except Exception as e:
            logger.error(f"Failed to load model {cache_key}: {e}")
            raise

unload_model async

unload_model(model_name: str)

Unload a model from memory. Accepts a catalog name or a composite base+LoRA cache key (A3.1).

Source code in adapta/core/model_manager.py

async def unload_model(self, model_name: str):
    """Unload a model from memory. Accepts a catalog name or a composite
    base+LoRA cache key (A3.1)."""
    if model_name in self._models:
        logger.info(f"Unloading model {model_name}")
        del self._models[model_name]
        self._infer_locks.pop(model_name, None)  # drop its serialization lock (A4.1)
        # The cache key may be a composite "name::lora::path"; only the base
        # catalog name has a config to flip back to unloaded.
        base = model_name.split("::lora::", 1)[0]
        if base in self._configs:
            self._configs[base].loaded = False

get_model

get_model(model_name: str) -> Optional[Llama]

Return a loaded model by cache key, or None if not loaded.

Source code in adapta/core/model_manager.py

def get_model(self, model_name: str) -> Optional[Llama]:
    """Return a loaded model by cache key, or None if not loaded."""
    return self._models.get(model_name)

list_models

list_models() -> list[ModelConfig]

List all available models with updated paths and existence status

Source code in adapta/core/model_manager.py

def list_models(self) -> list[ModelConfig]:
    """List all available models with updated paths and existence status"""
    models = []
    for config in self._configs.values():
        # Create a copy to avoid modifying the original
        model_copy = ModelConfig(
            name=config.name,
            model_type=config.model_type,
            path=config.path,
            context_length=config.context_length,
            n_threads=config.n_threads,
            n_gpu_layers=config.n_gpu_layers,
            description=config.description,
            loaded=config.loaded,
        )

        # Check if configured path exists, otherwise search for alternatives
        if not model_copy.path.exists():
            model_dir = model_copy.path.parent
            preferred_filename = model_copy.path.name
            found_path = self._find_model_file(model_dir, preferred_filename)
            if found_path:
                model_copy.path = found_path

        models.append(model_copy)
    return models

get_model_config

get_model_config(model_name: str) -> Optional[ModelConfig]

Get model configuration

Source code in adapta/core/model_manager.py

def get_model_config(self, model_name: str) -> Optional[ModelConfig]:
    """Get model configuration"""
    return self._configs.get(model_name)

get_model_by_type

get_model_by_type(model_type: ModelType) -> Optional[str]

Get the first available model of a given type

Source code in adapta/core/model_manager.py

def get_model_by_type(self, model_type: ModelType) -> Optional[str]:
    """Get the first available model of a given type"""
    for name, config in self._configs.items():
        if config.model_type == model_type:
            # Check if file exists, or try to find an alternative
            if config.path.exists():
                return name
            else:
                found_path = self._find_model_file(config.path.parent, config.path.name)
                if found_path:
                    return name
    return None

ensure_model_loaded async

ensure_model_loaded(
    model_name: str, adapter_path: Optional[str] = None
) -> Llama

Return the loaded model, loading it first if absent.

The cache_key not in _models check is not atomic with the subsequent load_model call. Two concurrent callers can both see the key absent and both call load_model — the second completes inside _load_lock and safely overwrites the first. The double-load is harmless but visible as two consecutive log lines.

Source code in adapta/core/model_manager.py

async def ensure_model_loaded(
    self, model_name: str, adapter_path: Optional[str] = None
) -> Llama:
    """Return the loaded model, loading it first if absent.

    The ``cache_key not in _models`` check is not atomic with the subsequent
    ``load_model`` call.  Two concurrent callers can both see the key absent
    and both call ``load_model`` — the second completes inside ``_load_lock``
    and safely overwrites the first.  The double-load is harmless but visible
    as two consecutive log lines.
    """
    cache_key = self._cache_key(self._resolve_serving_name(model_name), adapter_path)
    if cache_key not in self._models:
        return await self.load_model(model_name, adapter_path=adapter_path)
    self._models.move_to_end(cache_key)  # mark most-recently-used (A4.8)
    return self._models[cache_key]

is_loaded

is_loaded(
    model_name: str, adapter_path: Optional[str] = None
) -> bool

Check if a model (optionally a specific base+LoRA variant) is loaded

Source code in adapta/core/model_manager.py

def is_loaded(self, model_name: str, adapter_path: Optional[str] = None) -> bool:
    """Check if a model (optionally a specific base+LoRA variant) is loaded"""
    return self._cache_key(self._resolve_serving_name(model_name), adapter_path) in self._models

preload_default_models async

preload_default_models()

Preload default models (chat model)

Source code in adapta/core/model_manager.py

async def preload_default_models(self):
    """Preload default models (chat model)"""
    default_model = settings.default_model
    if default_model in self._configs:
        try:
            await self.load_model(default_model)
            logger.info(f"Preloaded default model: {default_model}")
        except Exception as e:
            logger.warning(f"Could not preload default model: {e}")