Skip to content

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 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@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 
41
42
43
44
45
46
47
48
49
50
51
52
@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)

InvalidAdapter dataclass

Bases: InvalidRequest

An imported adapter failed static validation (HTTP 400, code invalid_adapter).

The message always names the expected value (extension, base model, member list, ...).

Source code in adapta/domain/errors.py 
55
56
57
58
59
60
61
62
@dataclass
class InvalidAdapter(InvalidRequest):
    """An imported adapter failed static validation (HTTP 400, code ``invalid_adapter``).

    The message always names the expected value (extension, base model, member list, ...).
    """

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

AdapterIncompatible dataclass

Bases: InvalidAdapter

The adapter converts but cannot be loaded or run on its base model (HTTP 400, code adapter_incompatible): wrong rank/shape/target modules for the project's base.

Source code in adapta/domain/errors.py 
65
66
67
68
69
70
@dataclass
class AdapterIncompatible(InvalidAdapter):
    """The adapter converts but cannot be loaded or run on its base model (HTTP 400,
    code ``adapter_incompatible``): wrong rank/shape/target modules for the project's base."""

    code: str = field(default="adapter_incompatible", 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 
73
74
75
76
77
78
79
80
81
82
@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 
85
86
87
88
89
90
91
92
93
94
95
@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 
 98
 99
100
101
102
103
@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 
106
107
108
109
110
111
112
113
114
115
@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)

AdapterInUse dataclass

Bases: Conflict

An adapter cannot be deleted while an endpoint is bound to it (HTTP 409).

Source code in adapta/domain/errors.py 
118
119
120
121
122
@dataclass
class AdapterInUse(Conflict):
    """An adapter cannot be deleted while an endpoint is bound to it (HTTP 409)."""

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

DuplicateDocument dataclass

Bases: Conflict

Every chunk of an uploaded document already exists in the project (HTTP 409).

Source code in adapta/domain/errors.py 
125
126
127
128
129
@dataclass
class DuplicateDocument(Conflict):
    """Every chunk of an uploaded document already exists in the project (HTTP 409)."""

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

VectorStoreFailed dataclass

Bases: DomainError

The vector store (Chroma) rejected or failed an operation (HTTP 502).

The Postgres side is left untouched when this is raised, so the caller can retry.

Source code in adapta/domain/errors.py 
132
133
134
135
136
137
138
139
140
@dataclass
class VectorStoreFailed(DomainError):
    """The vector store (Chroma) rejected or failed an operation (HTTP 502).

    The Postgres side is left untouched when this is raised, so the caller can retry.
    """

    code: str = field(default="vector_store_failed", init=False)
    status: int = field(default=502, 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 
143
144
145
146
147
148
149
150
151
@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 
154
155
156
157
158
@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)

PayloadTooLarge dataclass

Bases: DomainError

An upload exceeded the configured size cap (HTTP 413).

Raised by the shared streaming-upload guard (adapta/api/_upload.py) before the oversized body is allowed to reach disk.

Source code in adapta/domain/errors.py 
161
162
163
164
165
166
167
168
169
170
@dataclass
class PayloadTooLarge(DomainError):
    """An upload exceeded the configured size cap (HTTP 413).

    Raised by the shared streaming-upload guard (``adapta/api/_upload.py``)
    before the oversized body is allowed to reach disk.
    """

    code: str = field(default="payload_too_large", init=False)
    status: int = field(default=413, init=False)

EmbeddingModelMismatch dataclass

Bases: DomainError

The embedding model configured now differs from the one a collection was actually built with (HTTP 409).

ADAPTA_EMBEDDING_MODEL changing after documents were indexed would otherwise silently corrupt retrieval (query embedded in a different vector space than the stored chunks) or crash Chroma on a dimension mismatch. message must name the fix (re-index, or restore the prior env var) — this is a config-drift error an operator can act on.

Source code in adapta/domain/errors.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@dataclass
class EmbeddingModelMismatch(DomainError):
    """The embedding model configured now differs from the one a collection
    was actually built with (HTTP 409).

    ``ADAPTA_EMBEDDING_MODEL`` changing after documents were indexed would
    otherwise silently corrupt retrieval (query embedded in a different
    vector space than the stored chunks) or crash Chroma on a dimension
    mismatch. ``message`` must name the fix (re-index, or restore the prior
    env var) — this is a config-drift error an operator can act on.
    """

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

UnknownJobKind dataclass

Bases: DomainError

A worker dequeued a job whose kind has no registered handler (HTTP 500).

Not client-facing today (the worker records it on the job); typed so the failure is a stable code rather than a raw string.

Source code in adapta/domain/errors.py 
189
190
191
192
193
194
195
196
197
198
@dataclass
class UnknownJobKind(DomainError):
    """A worker dequeued a job whose ``kind`` has no registered handler (HTTP 500).

    Not client-facing today (the worker records it on the job); typed so the failure
    is a stable code rather than a raw string.
    """

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

InferenceFailed dataclass

Bases: DomainError

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

Source code in adapta/domain/errors.py 
201
202
203
204
205
206
@dataclass
class InferenceFailed(DomainError):
    """The llama-cpp 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 
209
210
211
212
213
214
@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 
217
218
219
220
221
222
223
224
225
226
@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

An adapter did not clear the eval gate (HTTP 422).

HTTP 422 signals a business-rule rejection, not a server error: the check ran successfully, but the adapter is not good enough to serve.

Source code in adapta/domain/errors.py 
229
230
231
232
233
234
235
236
237
238
@dataclass
class EvalGateFailed(DomainError):
    """An adapter did not clear the eval gate (HTTP 422).

    HTTP 422 signals a business-rule rejection, not a server error: the check
    ran successfully, but the adapter is not good enough to serve.
    """

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

UnsafeTeacherUrl dataclass

Bases: DomainError

A synthesis teacher URL points at a blocked destination (HTTP 422, code unsafe_teacher_url): link-local addresses (cloud metadata), the metadata hostname, or a host that does not resolve. Loopback and private ranges are allowed.

Source code in adapta/domain/errors.py 
241
242
243
244
245
246
247
248
@dataclass
class UnsafeTeacherUrl(DomainError):
    """A synthesis teacher URL points at a blocked destination (HTTP 422, code
    ``unsafe_teacher_url``): link-local addresses (cloud metadata), the metadata hostname,
    or a host that does not resolve. Loopback and private ranges are allowed."""

    code: str = field(default="unsafe_teacher_url", 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 
251
252
253
254
255
256
@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 
259
260
261
262
263
264
265
266
267
268
269
@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)

ModelTooLargeForMemory dataclass

Bases: DomainError

The requested model cannot fit the memory budget even with nothing else loaded (HTTP 503).

Raised before loading so the kernel OOM killer never has to decide; the message names the estimate and the budget, and how to raise it.

Source code in adapta/domain/errors.py 
272
273
274
275
276
277
278
279
280
281
@dataclass
class ModelTooLargeForMemory(DomainError):
    """The requested model cannot fit the memory budget even with nothing else loaded (HTTP 503).

    Raised before loading so the kernel OOM killer never has to decide; the message names the
    estimate and the budget, and how to raise it.
    """

    code: str = field(default="model_too_large_for_memory", init=False)
    status: int = field(default=503, 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 
284
285
286
287
288
289
290
291
292
293
294
@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 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
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-1.5b-instruct"  # §E1.2: must resolve to a commercial_use catalog entry
    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
    # Memory budget for resident models (F1.6), in MiB. The count cap above ignores size: a 3B
    # beside a 1.5B plus the embedder crosses a 4 GB container and the kernel kills the server.
    # 0 (default) = derive from the container's cgroup limit minus `model_memory_reserve_mb`;
    # no cgroup limit = count cap only. >0 = use exactly this many MiB for loaded models.
    model_memory_budget_mb: int = 0
    # MiB of the cgroup limit kept for the embedder, torch, Python and request buffers
    # (measured: ~0.8 GB anon RSS with the embedder and a 1.5B model loaded).
    model_memory_reserve_mb: int = 1024

    # Serving backend (E2.1). "llamacpp" serves everything in-process. "ollama" sends
    # adapter-less text requests for models in `ollama_model_map` to an Ollama host;
    # adapters, vision and unmapped models stay on llama-cpp. The map is catalog name ->
    # Ollama tag (Apache-licensed entries only); override with a JSON object in
    # ADAPTA_OLLAMA_MODEL_MAP.
    serving_backend: Literal["llamacpp", "ollama"] = "llamacpp"
    ollama_base_url: str = "http://ollama:11434"
    ollama_model_map: dict[str, str] = {
        "qwen2.5-0.5b-instruct": "qwen2.5:0.5b-instruct",
        "qwen2.5-1.5b-instruct": "qwen2.5:1.5b-instruct",
        "qwen2.5-7b-instruct": "qwen2.5:7b-instruct",
    }

    # RAG / embeddings
    # Multilingual by default (E3.6, user decision 2026-10-10): on the bilingual golden set it
    # scores recall@5 0.978 vs 0.844 for the English-only all-MiniLM-L6-v2. Changing it on a
    # deployment with indexed projects needs `scripts/reembed.py` first (OPERATIONS 6.5).
    embedding_model: str = "paraphrase-multilingual-MiniLM-L12-v2"
    chroma_host: str = "chroma"
    chroma_port: int = 8000
    rag_top_k: int = 5
    # (E3.5) Chat-time relevance floor; 0 disables. Applies to RetrievedChunk.score: the
    # cross-encoder sigmoid when the reranker is on, else the vector cosine similarity
    # (1 - distance). Keyword-only hits have no cosine and are exempt. RRF scores are rank
    # arithmetic (max ~0.033), never compared against this. Not applied inside retrieve(),
    # so the golden set measures the same candidates as before.
    rag_min_score: float = 0.2
    # Hybrid retrieval (D5): fetch top_k * N vector candidates from Chroma for the
    # dense leg of the RRF fusion. 4× keeps latency low on CPU.
    rag_hybrid_fetch_multiplier: int = 4
    # Sparse leg (E3.2): Postgres full-text search over chunks.tsv, top_k * N hits.
    rag_sparse_fetch_multiplier: int = 4
    # Sparse leg drops query lexemes found in more than this fraction of the project's
    # chunks (language-agnostic statistical stopwords; E3.2). Terms in <= 3 chunks are kept.
    rag_sparse_max_df: float = 0.5
    # Text-search config for the query side: simple | english | spanish. chunks.tsv is
    # generated with 'simple' (E3.1), so anything else only matches after E3.6 regenerates it.
    rag_fts_config: Literal["simple", "english", "spanish"] = "simple"
    # Cross-encoder reranker model (D5) — OFF by default since E3.6: the only small one,
    # ms-marco-MiniLM, is English-only and cut bilingual recall@5 from 0.978 to 0.889 (and
    # tripled latency). Opt in with e.g. "cross-encoder/ms-marco-MiniLM-L-6-v2" for an
    # English-only corpus; rag_timeout_seconds guards the whole path including reranking.
    rag_reranker_model: str = ""
    # 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
    # (E1.15) Per-request bound on every call to the Chroma server. chromadb 1.5.9 builds its
    # httpx session with timeout=None, so without this a deadlocked Chroma blocks the caller
    # forever (measured 2026-10-10). Must exceed the slowest healthy upsert of one file.
    chroma_timeout_seconds: int = 30
    # /health/deep marks Chroma "degraded" when its heartbeat takes at least this long.
    chroma_slow_ms: int = 1000
    # (E1.15) Deadline for ONE attempt of an `index` job in the worker (parse + OCR + embed +
    # upsert). A timeout is a transient failure: one retry, then the file ends `failed`.
    # Sized for the largest scanned PDF the OCR path accepts; lower it for text-only deployments.
    index_job_timeout_seconds: int = 900
    # (E5.1) Contextual retrieval (opt-in per project). The situating sentence is written by the
    # project's base model unless a teacher endpoint is set (same shape as dataset synthesis).
    contextual_teacher_base_url: Optional[str] = None
    contextual_teacher_model: Optional[str] = None
    contextual_max_tokens: int = 80
    # Characters of surrounding document shown to the model per chunk (whole doc when it fits).
    contextual_window_chars: int = 12000
    # (E5.4) Visual extraction (opt-in per project via ``extraction_schema``, queued ``extract`` job).
    # Apache-2.0 vision base from the catalog; set to ``qwen2.5-vl-3b-instruct`` for a faster,
    # non-commercial run. CPU cost is per page, so the page count of one job is capped.
    extract_model: str = "qwen2.5-vl-7b-instruct"
    extract_max_pages: int = 20
    extract_max_tokens: int = 1024
    # (F1.1) Deadline for the model calls of ONE page in an `extract` job (~45 s/page on CPU, so the
    # job's bound is this x pages). A hung vision generation ends the job `failed`, not the worker.
    extract_page_timeout_seconds: int = 300
    # Longest side of the rendered page image sent to the vision model (bounds image tokens).
    extract_image_side_px: int = 1344
    chunk_size: int = 512
    chunk_overlap: int = 64
    # OCR'd chunks whose mean line confidence is below this are flagged needs_review (E3.3b).
    ocr_review_threshold: float = 0.90
    # 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

    # Eval gate (E4.3). An adapter is scored on a held-out eval set through the serving path
    # (greedy generation, task metric) and compared with the base model on the SAME rows. It
    # passes if EITHER it beats the base by `eval_min_improvement` (it demonstrably helped) OR
    # it clears `eval_absolute_floor` and is at least as good as the base (already strong).
    eval_min_improvement: float = 0.05
    eval_absolute_floor: float = 0.8
    # An eval set must have this many rows: fewer is noise, more is wasted CPU time.
    eval_min_rows: int = 20
    eval_max_rows: int = 200
    # Generation cap per eval row (classification labels and JSON extractions are short).
    eval_max_new_tokens: int = 128

    # Hard cap on a single dataset UPLOAD, enforced while streaming to disk so a
    # hostile/oversized upload can't fill the disk before validation even runs.
    max_upload_mb: int = 1024

    # Cap on one adapter upload (a .gguf, or a .zip of a PEFT adapter) and on the size a zip
    # may expand to — LoRA adapters are tens to hundreds of MB; this bounds a zip bomb.
    adapter_max_upload_mb: int = 512

    # Image limits for vision chat (§V4): decoded size/dimensions per image, and
    # the number of image content-parts per request.
    max_image_mb: int = 10
    max_image_side_px: int = 8192
    max_images_per_request: int = 4

    # Adapter 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.
    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)

    # 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).

split_client_system 

split_client_system(
    messages: List[dict],
) -> tuple[List[dict], Optional[str]]

Pull system turns out of the conversation (E3.5).

Returns (remaining turns, their text joined as one paragraph). They are merged into the single system message after the project prompt and RAG block, instead of landing as a later turn the model weighs less than the retrieved context.

Source code in adapta/services/chat.py 
162
163
164
165
166
167
168
169
170
171
172
def split_client_system(messages: List[dict]) -> tuple[List[dict], Optional[str]]:
    """Pull `system` turns out of the conversation (E3.5).

    Returns (remaining turns, their text joined as one paragraph). They are merged into
    the single system message after the project prompt and RAG block, instead of landing
    as a later turn the model weighs less than the retrieved context.
    """
    turns = [m for m in messages if m.get("role") != "system"]
    texts = [flatten_text(m.get("content", "")).strip() for m in messages if m.get("role") == "system"]
    joined = "\n\n".join(t for t in texts if t)
    return turns, (joined or None)

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 
189
190
191
192
193
194
195
196
197
198
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 
201
202
203
204
205
206
207
208
209
210
211
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 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
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

is_abstention 

is_abstention(text: str) -> bool

True when the model's answer is the canonical abstention (prefix, case-insensitive).

Source code in adapta/services/chat.py 
328
329
330
def is_abstention(text: str) -> bool:
    """True when the model's answer is the canonical abstention (prefix, case-insensitive)."""
    return text.strip().lower().startswith(ABSTENTION_ANSWER.lower().rstrip("."))

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,
    rag_filters: Optional[RetrievalFilters] = None,
    adapter_path: Optional[str] = None,
    response_format: Optional[dict] = None,
    condense: bool = True
) -> dict

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

Source code in adapta/services/chat.py 
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
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,
    rag_filters: Optional[RetrievalFilters] = None,
    # Fine-tune serving (A3.1): GGUF LoRA applied on top of the base model
    adapter_path: Optional[str] = None,
    # JSON mode: OpenAI-shape response_format dict (None = unconstrained)
    response_format: Optional[dict] = None,
    # E5.3: condense multi-turn history into a standalone retrieval query (adapta.condense)
    condense: bool = True,
) -> 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,
            response_format=response_format,
        )

    messages, client_system = split_client_system(messages)
    rag_chunks: list = []
    rag_service = None
    backend: Any = None
    handle: Any = None
    cond_pt = cond_ct = 0
    if project_id:
        rag_service = get_rag_service()
        query, cond_pt, cond_ct, backend, handle = await _retrieval_queries(
            model_name=model_name, adapter_path=adapter_path, messages=messages,
            condense=condense,
        )
        effective_top_k = await _resolve_rag_top_k(project_id, top_k_rag)
        rag_chunks = await _retrieve(
            rag_service, project_id, query, effective_top_k, rag_filters
        )

    if project_id and not rag_chunks:
        canned = _abstention_response(model_name)
        canned["usage"] = {  # only the condensation call (if any) ran
            "prompt_tokens": cond_pt,
            "completion_tokens": cond_ct,
            "total_tokens": cond_pt + cond_ct,
        }
        return canned

    # 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
    ]

    if backend is None:
        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); 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,
            abstain=project_id is not None,
            client_system=client_system,
            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,
        response_format=response_format,
    )
    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": {
            # + the condensation call (E5.3): it is metered like any generation
            "prompt_tokens": response.prompt_tokens + cond_pt,
            "completion_tokens": response.completion_tokens + cond_ct,
            "total_tokens": (
                response.prompt_tokens + response.completion_tokens + cond_pt + cond_ct
            ),
        },
    }

    if project_id:  # [] when nothing cleared the threshold: the model was told to abstain
        rag_service = get_rag_service()
        result["citations"] = (
            [] if is_abstention(response.content) else 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,
    rag_filters: Optional[RetrievalFilters] = None,
    endpoint_id: Optional[str] = None,
    adapter_path: Optional[str] = None,
    response_format: Optional[dict] = None,
    condense: bool = True
) -> 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 
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
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,
    rag_filters: Optional[RetrievalFilters] = None,
    endpoint_id: Optional[str] = None,
    adapter_path: Optional[str] = None,
    response_format: Optional[dict] = None,
    condense: bool = True,
) -> 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."
        )

    messages, client_system = split_client_system(messages)
    rag_chunks: list = []
    rag_service = None
    backend: Any = None
    handle: Any = None
    cond_pt = cond_ct = 0
    if project_id:
        rag_service = get_rag_service()
        query, cond_pt, cond_ct, backend, handle = await _retrieval_queries(
            model_name=model_name, adapter_path=adapter_path, messages=messages,
            condense=condense,
        )
        effective_top_k = await _resolve_rag_top_k(project_id, top_k_rag)
        rag_chunks = await _retrieve(
            rag_service, project_id, query, effective_top_k, rag_filters
        )

    if project_id and not rag_chunks:
        canned = _abstention_response(model_name)
        canned["usage"] = {
            "prompt_tokens": cond_pt,
            "completion_tokens": cond_ct,
            "total_tokens": cond_pt + cond_ct,
        }
        frame = {
            "id": canned["id"],
            "object": "chat.completion.chunk",
            "created": canned["created"],
            "model": model_name,
        }
        content = {"index": 0, "delta": {"content": ABSTENTION_ANSWER}, "finish_reason": None}
        yield f"data: {json.dumps({**frame, 'choices': [content]})}\n\n"
        final = {**frame, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}
        final["usage"] = canned["usage"]
        final["citations"] = []
        yield f"data: {json.dumps(final)}\n\n"
        yield "data: [DONE]\n\n"
        if endpoint_id:
            from adapta.services.usage import record_usage

            await record_usage(endpoint_id, cond_pt, cond_ct)
        return

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

    # Backend selection (D3): same routing as non-streaming chat().
    if backend is None:
        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,
        abstain=project_id is not None,
        client_system=client_system,
        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,
        response_format=response_format,
    )
    chunk_id = f"chatcmpl-{uuid.uuid4().hex[:8]}"
    created = int(time.time())

    # llama-cpp: exact tokenizer count.
    prompt_tokens = handle.count_prompt_tokens(inference_messages, full_system)
    # Each iteration is one generated token, so counting iterations is exact.
    completion_tokens = 0
    answer_parts: list[str] = []
    try:
        async for token in backend.generate_stream(handle, req):
            completion_tokens += 1
            answer_parts.append(token)
            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"

        engine_usage = handle.usage
        if engine_usage:  # engine-reported exact counts (Ollama) beat our own
            prompt_tokens, completion_tokens = engine_usage
        prompt_tokens += cond_pt  # the condensation call (E5.3) is metered too
        completion_tokens += cond_ct
        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,
            },
        }
        # Citations can only be known once retrieval has happened — same
        # shape as chat()'s non-streaming response, carried on the final
        # frame (E1.7) since the model prompt (with [N] markers) is what
        # the SSE tokens are being streamed for.
        if project_id:
            if rag_service is None:
                raise InvalidRequest(message="RAG service not configured")
            finish["citations"] = (
                [] if is_abstention("".join(answer_parts)) else rag_service.format_citations(rag_chunks)
            )
        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. Dense leg — top top_k × multiplier candidates from Chroma. 2. Sparse leg — Postgres full-text search over chunks.tsv (whole corpus, E3.2). 3. Reciprocal Rank Fusion of the two independent ranked lists. 4. Optional cross-encoder rerank over the top candidates. 5. Return top_k chunks, best-first (§A4.3 pop() invariant preserved).

RetrievalFilters dataclass

(E3.5) Request-level restriction of retrieval. All fields optional (AND-ed).

Source code in adapta/services/rag.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@dataclass(frozen=True)
class RetrievalFilters:
    """(E3.5) Request-level restriction of retrieval. All fields optional (AND-ed)."""

    file_ids: Optional[List[str]] = None
    page_min: Optional[int] = None
    page_max: Optional[int] = None
    needs_review: Optional[bool] = None
    # (E5.4) Exact match on extracted fields (``chunks.fields ->> key``), AND-ed.
    fields: Optional[dict] = None

    @property
    def sql_only(self) -> bool:
        """Filters Chroma metadata cannot express (it carries no page / review flag / fields)."""
        return (
            self.page_min is not None
            or self.page_max is not None
            or self.needs_review is not None
            or bool(self.fields)
        )

    def sql_clauses(self) -> tuple[str, dict]:
        """``AND ...`` fragment over alias ``c`` (chunks) plus its bind params."""
        parts: list[str] = []
        params: dict = {}
        if self.file_ids:
            parts.append("c.file_id = ANY(:f_file_ids)")
            params["f_file_ids"] = list(self.file_ids)
        if self.page_min is not None:
            parts.append("c.page >= :f_page_min")
            params["f_page_min"] = self.page_min
        if self.page_max is not None:
            parts.append("c.page <= :f_page_max")
            params["f_page_max"] = self.page_max
        if self.needs_review is not None:
            parts.append("c.needs_review = :f_needs_review")
            params["f_needs_review"] = self.needs_review
        for i, (key, value) in enumerate((self.fields or {}).items()):
            parts.append(f"c.fields ->> :f_fk{i} = :f_fv{i}")
            params[f"f_fk{i}"] = str(key)
            params[f"f_fv{i}"] = str(value)
        return "".join(f" AND {p}" for p in parts), params
sql_only property 
sql_only: bool

Filters Chroma metadata cannot express (it carries no page / review flag / fields).

sql_clauses 
sql_clauses() -> tuple[str, dict]

AND ... fragment over alias c (chunks) plus its bind params.

Source code in adapta/services/rag.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def sql_clauses(self) -> tuple[str, dict]:
    """``AND ...`` fragment over alias ``c`` (chunks) plus its bind params."""
    parts: list[str] = []
    params: dict = {}
    if self.file_ids:
        parts.append("c.file_id = ANY(:f_file_ids)")
        params["f_file_ids"] = list(self.file_ids)
    if self.page_min is not None:
        parts.append("c.page >= :f_page_min")
        params["f_page_min"] = self.page_min
    if self.page_max is not None:
        parts.append("c.page <= :f_page_max")
        params["f_page_max"] = self.page_max
    if self.needs_review is not None:
        parts.append("c.needs_review = :f_needs_review")
        params["f_needs_review"] = self.needs_review
    for i, (key, value) in enumerate((self.fields or {}).items()):
        parts.append(f"c.fields ->> :f_fk{i} = :f_fv{i}")
        params[f"f_fk{i}"] = str(key)
        params[f"f_fv{i}"] = str(value)
    return "".join(f" AND {p}" for p in parts), params

RAGService

Per-project ChromaDB collection management and hybrid cited retrieval.

Every call goes through the one process-wide Chroma client (_client() → _get_chroma_client()); this object holds no connection itself.

Source code in adapta/services/rag.py 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
class RAGService:
    """Per-project ChromaDB collection management and hybrid cited retrieval.

    Every call goes through the one process-wide Chroma client
    (``_client()`` → ``_get_chroma_client()``); this object holds no connection itself.
    """

    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, and
        records the embedding model active right now as ``embedding_model``.
        This metadata is applied only at creation — Chroma ignores it on
        ``get_or_create_collection`` calls for an existing collection, so both
        the distance metric and the embedding-model record are locked in
        after the first call. ``retrieve()`` reads the latter back to detect
        ``ADAPTA_EMBEDDING_MODEL`` drift.
        """
        name = collection_name_for(project_id)
        with _vector_store("create the project collection"):
            self._client().get_or_create_collection(
                name=name,
                metadata={"hnsw:space": "cosine", "embedding_model": settings.embedding_model},
            )
        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)
        try:
            self._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.

        Vectors only — the Postgres rows are written by ``persist_chunks`` before this.

        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]
        # (E5.1) The vector covers context + text; the stored document stays the raw text so
        # citations quote the source. No context -> the exact pre-E5.1 input.
        embeddings = emb_service.embed([indexed_text(c) for c in chunks])

        name = collection_name_for(project_id)
        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
        ]
        with _vector_store("store the file's vectors"):
            collection = self._client().get_or_create_collection(
                name=name,
                metadata={"hnsw:space": "cosine", "embedding_model": settings.embedding_model},
            )
            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 vectors for a file from the Chroma collection.

        Raises ``VectorStoreFailed`` on any Chroma error (the Postgres rows go by CASCADE).

        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)
        try:
            collection = self._client().get_collection(name)
            results = collection.get(where={"file_id": file_id})
            if results["ids"]:
                collection.delete(ids=results["ids"])
        except Exception as exc:
            raise VectorStoreFailed(
                message="Could not remove the file's vectors from the index; nothing was deleted.",
                internal_detail=f"chroma delete failed for file {file_id}: {exc!r}",
            ) from exc

    def delete_chunk_ids(self, project_id: str, ids: List[str]) -> None:
        """(E5.4) Remove specific vectors (ids are ``{file_id}_{index}``); raises ``VectorStoreFailed``."""
        if not ids:
            return
        with _vector_store("remove replaced chunk vectors"):
            self._client().get_collection(collection_name_for(project_id)).delete(ids=ids)

    def retrieve(
        self,
        project_id: str,
        query: str | Sequence[str],
        top_k: Optional[int] = None,
        filters: Optional[RetrievalFilters] = None,
    ) -> List[RetrievedChunk]:
        """Hybrid retrieval: Chroma vector leg + Postgres FTS leg fused via RRF, then rerank.

        Pipeline:
          1. Dense: fetch top_k × rag_hybrid_fetch_multiplier candidates from Chroma.
          2. Sparse: Postgres FTS over chunks.tsv, top_k × rag_sparse_fetch_multiplier.
          3. RRF over the two independent ranked lists (union by chunk id).
          4. Optional cross-encoder reranker over the top rerank_n candidates.
          5. Return top_k, best-first (§A4.3 pop() invariant preserved); log one line.

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

        ``query`` is one string or a list of queries (E5.3: the condensed standalone question
        plus the original). Each query runs both legs; one RRF fuses all the lists. The first
        query is the primary one (the reranker scores against it). A plain string is the
        single-query case, unchanged.

        ``filters`` (E3.5) restricts both legs: ``file_ids`` as a Chroma ``where`` and SQL
        ``WHERE`` on the FTS leg; page / review filters in SQL on the FTS leg and over the
        vector candidates (over-fetched 4x to leave survivors). ``None`` changes nothing.
        """
        t0 = time.perf_counter()
        query_list = [query] if isinstance(query, str) else list(query)
        top_k = top_k or settings.rag_top_k
        fetch_k = max(top_k * settings.rag_hybrid_fetch_multiplier, top_k + 1)
        if filters is not None and filters.sql_only:
            fetch_k *= 4

        name = collection_name_for(project_id)
        client = self._client()
        try:
            collection = client.get_collection(name)
        except Exception as exc:
            logger.warning("retrieve: no usable Chroma collection %s (%r); returning no hits", name, exc)
            return []

        # Config-drift guard: the collection's Chroma metadata records the
        # embedding model it was actually built with (set once, at creation —
        # see ensure_collection). If ADAPTA_EMBEDDING_MODEL changed since, a
        # query embedded with the new model would search the wrong vector
        # space — or blow up Chroma on a dimension mismatch — either way not
        # a signal a caller should have to decode from a Chroma error. `None`
        # (a collection indexed before this guard existed) is unknown
        # provenance, not treated as a mismatch.
        stored_model = (getattr(collection, "metadata", None) or {}).get("embedding_model")
        if stored_model and stored_model != settings.embedding_model:
            raise EmbeddingModelMismatch(
                message=(
                    f"This project's documents were indexed with '{stored_model}' "
                    f"but the server is now configured for '{settings.embedding_model}'. "
                    f"Re-index the project's files, or restore ADAPTA_EMBEDDING_MODEL={stored_model}."
                )
            )

        emb_service = get_embedding_service()
        query_kwargs: dict = {}
        if filters is not None and filters.file_ids:
            query_kwargs["where"] = {"file_id": {"$in": list(filters.file_ids)}}

        # One vector list and one FTS list per query (E5.3). A chunk seen under several
        # queries keeps its best cosine score; RRF below fuses every list at once.
        by_id: dict[str, RetrievedChunk] = {}
        vec_lists: list[list[str]] = []
        fts_lists: list[list[str]] = []
        n_vec = n_fts = 0
        for q in query_list:
            q_emb = emb_service.embed_one(q)
            results = collection.query(
                query_embeddings=[q_emb],
                n_results=fetch_k,
                include=["documents", "metadatas", "distances"],
                **query_kwargs,
            )
            # Chroma leaves exact-distance ties in arbitrary order: pin them by content (F1.4).
            order = sorted(
                range(len(results["ids"][0])),
                key=lambda j: (
                    results["distances"][0][j],
                    _chunk_tie_key(results["metadatas"][0][j], results["ids"][0][j]),
                ),
            )
            ids: list[str] = [results["ids"][0][j] for j in order]
            docs: list[str] = [results["documents"][0][j] for j in order]
            metas: list[dict] = [results["metadatas"][0][j] for j in order]
            dists: list[float] = [results["distances"][0][j] for j in order]
            if filters is not None and filters.sql_only:
                keep = _filter_vector_ids(ids, filters)
                kept = [i for i, cid in enumerate(ids) if cid in keep]
                ids = [ids[i] for i in kept]
                docs = [docs[i] for i in kept]
                metas = [metas[i] for i in kept]
                dists = [dists[i] for i in kept]
            for i, cid in enumerate(ids):
                score = 1.0 - dists[i]  # cosine distance -> similarity
                prev = by_id.get(cid)
                if prev is None or score > prev.score:
                    by_id[cid] = RetrievedChunk(
                        text=docs[i],
                        source=metas[i].get("source", "unknown"),
                        score=score,
                        chunk_index=metas[i].get("chunk_index", 0),
                        score_kind="vector",
                        chunk_id=cid,
                    )
            vec_lists.append(ids)
            n_vec += len(ids)

            # Sparse leg over the whole corpus. A Postgres failure degrades to vector-only
            # rather than failing the request (the dense leg is still a valid answer).
            try:
                fts_kwargs = {"filters": filters} if filters is not None else {}
                fts_hits = _fts_search(
                    project_id, q, top_k * settings.rag_sparse_fetch_multiplier, **fts_kwargs
                )
            except (SQLAlchemyError, OSError, asyncio.TimeoutError) as exc:
                logger.warning("rag fts leg failed, vector-only: %r", exc)
                fts_hits = []
            for h in fts_hits:
                # Keyword-only hit the embedder missed: no cosine score exists for it.
                by_id.setdefault(
                    h.id,
                    RetrievedChunk(
                        text=h.text, source=h.source, score=0.0,
                        chunk_index=h.chunk_index, score_kind="vector", chunk_id=h.id,
                        lexical_only=True,
                    ),
                )
            fts_lists.append([h.id for h in fts_hits])
            n_fts += len(fts_hits)

        if not by_id:
            return []

        fused_ids = _rrf_fuse(
            *vec_lists,
            *fts_lists,
            tie_key={cid: _chunk_tie_key({"source": c.source, "chunk_index": c.chunk_index}, cid)
                     for cid, c in by_id.items()},
        )
        candidates: List[RetrievedChunk] = [by_id[cid] for cid in fused_ids]

        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]
            t_rerank = time.perf_counter()
            raw_scores = reranker.rerank(query_list[0], [c.text for c in to_rerank])
            get_metrics_collector().rag_reranker_seconds.observe(time.perf_counter() - t_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,
                    score_kind="reranker",
                    chunk_id=c.chunk_id,
                )
                for s, c in ranked
            ]
            candidates = reranked + candidates[rerank_n:]

        final = _hydrate_provenance(candidates[:top_k])
        fused_rank = {cid: r for r, cid in enumerate(fused_ids)}
        mc = get_metrics_collector()
        mc.rag_retrieval_seconds.observe(time.perf_counter() - t0)
        mc.rag_chunks_returned.observe(len(final))
        logger.info(
            "rag.retrieve q=%s project=%s vec=%d fts=%d fused=%d ids=%s scores=%s ms=%d",
            "+".join(hashlib.sha256(q.encode("utf-8")).hexdigest()[:12] for q in query_list),
            project_id,
            n_vec,
            n_fts,
            len(fused_ids),
            [f"{c.chunk_id}@{fused_rank[c.chunk_id]}" for c in final],
            [round(c.score, 4) for c in final],
            int((time.perf_counter() - t0) * 1000),
        )
        return final

    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):
            head = f"{c.fields_header}\n" if c.fields_header else ""
            parts.append(f"[{i}] (source: {c.source})\n{head}{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),
                "score_kind": c.score_kind,
                "text": c.text,
                "file_id": c.file_id,
                "page": c.page,
                "ocr": c.ocr,
                "confidence": c.ocr_confidence,
            }
            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, and records the embedding model active right now as embedding_model. This metadata is applied only at creation — Chroma ignores it on get_or_create_collection calls for an existing collection, so both the distance metric and the embedding-model record are locked in after the first call. retrieve() reads the latter back to detect ADAPTA_EMBEDDING_MODEL drift.

Source code in adapta/services/rag.py 
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
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, and
    records the embedding model active right now as ``embedding_model``.
    This metadata is applied only at creation — Chroma ignores it on
    ``get_or_create_collection`` calls for an existing collection, so both
    the distance metric and the embedding-model record are locked in
    after the first call. ``retrieve()`` reads the latter back to detect
    ``ADAPTA_EMBEDDING_MODEL`` drift.
    """
    name = collection_name_for(project_id)
    with _vector_store("create the project collection"):
        self._client().get_or_create_collection(
            name=name,
            metadata={"hnsw:space": "cosine", "embedding_model": settings.embedding_model},
        )
    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 
526
527
528
529
530
531
532
533
534
535
536
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)
    try:
        self._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.

Vectors only — the Postgres rows are written by persist_chunks before this.

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 
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
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.

    Vectors only — the Postgres rows are written by ``persist_chunks`` before this.

    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]
    # (E5.1) The vector covers context + text; the stored document stays the raw text so
    # citations quote the source. No context -> the exact pre-E5.1 input.
    embeddings = emb_service.embed([indexed_text(c) for c in chunks])

    name = collection_name_for(project_id)
    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
    ]
    with _vector_store("store the file's vectors"):
        collection = self._client().get_or_create_collection(
            name=name,
            metadata={"hnsw:space": "cosine", "embedding_model": settings.embedding_model},
        )
        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 vectors for a file from the Chroma collection.

Raises VectorStoreFailed on any Chroma error (the Postgres rows go by CASCADE).

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 
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
def delete_file_chunks(self, project_id: str, file_id: str) -> None:
    """Remove all vectors for a file from the Chroma collection.

    Raises ``VectorStoreFailed`` on any Chroma error (the Postgres rows go by CASCADE).

    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)
    try:
        collection = self._client().get_collection(name)
        results = collection.get(where={"file_id": file_id})
        if results["ids"]:
            collection.delete(ids=results["ids"])
    except Exception as exc:
        raise VectorStoreFailed(
            message="Could not remove the file's vectors from the index; nothing was deleted.",
            internal_detail=f"chroma delete failed for file {file_id}: {exc!r}",
        ) from exc
delete_chunk_ids 
delete_chunk_ids(project_id: str, ids: List[str]) -> None

(E5.4) Remove specific vectors (ids are {file_id}_{index}); raises VectorStoreFailed.

Source code in adapta/services/rag.py 
593
594
595
596
597
598
def delete_chunk_ids(self, project_id: str, ids: List[str]) -> None:
    """(E5.4) Remove specific vectors (ids are ``{file_id}_{index}``); raises ``VectorStoreFailed``."""
    if not ids:
        return
    with _vector_store("remove replaced chunk vectors"):
        self._client().get_collection(collection_name_for(project_id)).delete(ids=ids)
retrieve 
retrieve(
    project_id: str,
    query: str | Sequence[str],
    top_k: Optional[int] = None,
    filters: Optional[RetrievalFilters] = None,
) -> List[RetrievedChunk]

Hybrid retrieval: Chroma vector leg + Postgres FTS leg fused via RRF, then rerank.

Pipeline
  1. Dense: fetch top_k × rag_hybrid_fetch_multiplier candidates from Chroma.
  2. Sparse: Postgres FTS over chunks.tsv, top_k × rag_sparse_fetch_multiplier.
  3. RRF over the two independent ranked lists (union by chunk id).
  4. Optional cross-encoder reranker over the top rerank_n candidates.
  5. Return top_k, best-first (§A4.3 pop() invariant preserved); log one line.

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

query is one string or a list of queries (E5.3: the condensed standalone question plus the original). Each query runs both legs; one RRF fuses all the lists. The first query is the primary one (the reranker scores against it). A plain string is the single-query case, unchanged.

filters (E3.5) restricts both legs: file_ids as a Chroma where and SQL WHERE on the FTS leg; page / review filters in SQL on the FTS leg and over the vector candidates (over-fetched 4x to leave survivors). None changes nothing.

Source code in adapta/services/rag.py 
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
def retrieve(
    self,
    project_id: str,
    query: str | Sequence[str],
    top_k: Optional[int] = None,
    filters: Optional[RetrievalFilters] = None,
) -> List[RetrievedChunk]:
    """Hybrid retrieval: Chroma vector leg + Postgres FTS leg fused via RRF, then rerank.

    Pipeline:
      1. Dense: fetch top_k × rag_hybrid_fetch_multiplier candidates from Chroma.
      2. Sparse: Postgres FTS over chunks.tsv, top_k × rag_sparse_fetch_multiplier.
      3. RRF over the two independent ranked lists (union by chunk id).
      4. Optional cross-encoder reranker over the top rerank_n candidates.
      5. Return top_k, best-first (§A4.3 pop() invariant preserved); log one line.

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

    ``query`` is one string or a list of queries (E5.3: the condensed standalone question
    plus the original). Each query runs both legs; one RRF fuses all the lists. The first
    query is the primary one (the reranker scores against it). A plain string is the
    single-query case, unchanged.

    ``filters`` (E3.5) restricts both legs: ``file_ids`` as a Chroma ``where`` and SQL
    ``WHERE`` on the FTS leg; page / review filters in SQL on the FTS leg and over the
    vector candidates (over-fetched 4x to leave survivors). ``None`` changes nothing.
    """
    t0 = time.perf_counter()
    query_list = [query] if isinstance(query, str) else list(query)
    top_k = top_k or settings.rag_top_k
    fetch_k = max(top_k * settings.rag_hybrid_fetch_multiplier, top_k + 1)
    if filters is not None and filters.sql_only:
        fetch_k *= 4

    name = collection_name_for(project_id)
    client = self._client()
    try:
        collection = client.get_collection(name)
    except Exception as exc:
        logger.warning("retrieve: no usable Chroma collection %s (%r); returning no hits", name, exc)
        return []

    # Config-drift guard: the collection's Chroma metadata records the
    # embedding model it was actually built with (set once, at creation —
    # see ensure_collection). If ADAPTA_EMBEDDING_MODEL changed since, a
    # query embedded with the new model would search the wrong vector
    # space — or blow up Chroma on a dimension mismatch — either way not
    # a signal a caller should have to decode from a Chroma error. `None`
    # (a collection indexed before this guard existed) is unknown
    # provenance, not treated as a mismatch.
    stored_model = (getattr(collection, "metadata", None) or {}).get("embedding_model")
    if stored_model and stored_model != settings.embedding_model:
        raise EmbeddingModelMismatch(
            message=(
                f"This project's documents were indexed with '{stored_model}' "
                f"but the server is now configured for '{settings.embedding_model}'. "
                f"Re-index the project's files, or restore ADAPTA_EMBEDDING_MODEL={stored_model}."
            )
        )

    emb_service = get_embedding_service()
    query_kwargs: dict = {}
    if filters is not None and filters.file_ids:
        query_kwargs["where"] = {"file_id": {"$in": list(filters.file_ids)}}

    # One vector list and one FTS list per query (E5.3). A chunk seen under several
    # queries keeps its best cosine score; RRF below fuses every list at once.
    by_id: dict[str, RetrievedChunk] = {}
    vec_lists: list[list[str]] = []
    fts_lists: list[list[str]] = []
    n_vec = n_fts = 0
    for q in query_list:
        q_emb = emb_service.embed_one(q)
        results = collection.query(
            query_embeddings=[q_emb],
            n_results=fetch_k,
            include=["documents", "metadatas", "distances"],
            **query_kwargs,
        )
        # Chroma leaves exact-distance ties in arbitrary order: pin them by content (F1.4).
        order = sorted(
            range(len(results["ids"][0])),
            key=lambda j: (
                results["distances"][0][j],
                _chunk_tie_key(results["metadatas"][0][j], results["ids"][0][j]),
            ),
        )
        ids: list[str] = [results["ids"][0][j] for j in order]
        docs: list[str] = [results["documents"][0][j] for j in order]
        metas: list[dict] = [results["metadatas"][0][j] for j in order]
        dists: list[float] = [results["distances"][0][j] for j in order]
        if filters is not None and filters.sql_only:
            keep = _filter_vector_ids(ids, filters)
            kept = [i for i, cid in enumerate(ids) if cid in keep]
            ids = [ids[i] for i in kept]
            docs = [docs[i] for i in kept]
            metas = [metas[i] for i in kept]
            dists = [dists[i] for i in kept]
        for i, cid in enumerate(ids):
            score = 1.0 - dists[i]  # cosine distance -> similarity
            prev = by_id.get(cid)
            if prev is None or score > prev.score:
                by_id[cid] = RetrievedChunk(
                    text=docs[i],
                    source=metas[i].get("source", "unknown"),
                    score=score,
                    chunk_index=metas[i].get("chunk_index", 0),
                    score_kind="vector",
                    chunk_id=cid,
                )
        vec_lists.append(ids)
        n_vec += len(ids)

        # Sparse leg over the whole corpus. A Postgres failure degrades to vector-only
        # rather than failing the request (the dense leg is still a valid answer).
        try:
            fts_kwargs = {"filters": filters} if filters is not None else {}
            fts_hits = _fts_search(
                project_id, q, top_k * settings.rag_sparse_fetch_multiplier, **fts_kwargs
            )
        except (SQLAlchemyError, OSError, asyncio.TimeoutError) as exc:
            logger.warning("rag fts leg failed, vector-only: %r", exc)
            fts_hits = []
        for h in fts_hits:
            # Keyword-only hit the embedder missed: no cosine score exists for it.
            by_id.setdefault(
                h.id,
                RetrievedChunk(
                    text=h.text, source=h.source, score=0.0,
                    chunk_index=h.chunk_index, score_kind="vector", chunk_id=h.id,
                    lexical_only=True,
                ),
            )
        fts_lists.append([h.id for h in fts_hits])
        n_fts += len(fts_hits)

    if not by_id:
        return []

    fused_ids = _rrf_fuse(
        *vec_lists,
        *fts_lists,
        tie_key={cid: _chunk_tie_key({"source": c.source, "chunk_index": c.chunk_index}, cid)
                 for cid, c in by_id.items()},
    )
    candidates: List[RetrievedChunk] = [by_id[cid] for cid in fused_ids]

    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]
        t_rerank = time.perf_counter()
        raw_scores = reranker.rerank(query_list[0], [c.text for c in to_rerank])
        get_metrics_collector().rag_reranker_seconds.observe(time.perf_counter() - t_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,
                score_kind="reranker",
                chunk_id=c.chunk_id,
            )
            for s, c in ranked
        ]
        candidates = reranked + candidates[rerank_n:]

    final = _hydrate_provenance(candidates[:top_k])
    fused_rank = {cid: r for r, cid in enumerate(fused_ids)}
    mc = get_metrics_collector()
    mc.rag_retrieval_seconds.observe(time.perf_counter() - t0)
    mc.rag_chunks_returned.observe(len(final))
    logger.info(
        "rag.retrieve q=%s project=%s vec=%d fts=%d fused=%d ids=%s scores=%s ms=%d",
        "+".join(hashlib.sha256(q.encode("utf-8")).hexdigest()[:12] for q in query_list),
        project_id,
        n_vec,
        n_fts,
        len(fused_ids),
        [f"{c.chunk_id}@{fused_rank[c.chunk_id]}" for c in final],
        [round(c.score, 4) for c in final],
        int((time.perf_counter() - t0) * 1000),
    )
    return final
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 
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
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):
        head = f"{c.fields_header}\n" if c.fields_header else ""
        parts.append(f"[{i}] (source: {c.source})\n{head}{c.text}")
    return "\n\n---\n\n".join(parts)

apply_min_score 

apply_min_score(
    chunks: List["RetrievedChunk"], min_score: float
) -> List["RetrievedChunk"]

Drop chunks whose relevance score is under min_score (0 disables).

score is the reranker sigmoid or the vector cosine, depending on score_kind — both in [0, 1] and both thresholded as-is. Keyword-only hits are kept: FTS already discards statistical stopwords, so a hit is exact-term evidence the cosine cannot judge.

Source code in adapta/services/rag.py 
106
107
108
109
110
111
112
113
114
115
def apply_min_score(chunks: List["RetrievedChunk"], min_score: float) -> List["RetrievedChunk"]:
    """Drop chunks whose relevance score is under ``min_score`` (0 disables).

    ``score`` is the reranker sigmoid or the vector cosine, depending on ``score_kind`` —
    both in [0, 1] and both thresholded as-is. Keyword-only hits are kept: FTS already
    discards statistical stopwords, so a hit is exact-term evidence the cosine cannot judge.
    """
    if min_score <= 0:
        return chunks
    return [c for c in chunks if c.lexical_only or c.score >= min_score]

content_hash 

content_hash(text: str) -> str

sha256 hex digest of a chunk's text — the dedup key (64 chars).

Source code in adapta/services/rag.py 
118
119
120
def content_hash(text: str) -> str:
    """sha256 hex digest of a chunk's text — the dedup key (64 chars)."""
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

indexed_text 

indexed_text(chunk: object) -> str

What gets embedded for a chunk: context + text when it has a situating sentence.

Source code in adapta/services/rag.py 
130
131
132
133
134
def indexed_text(chunk: object) -> str:
    """What gets embedded for a chunk: ``context + text`` when it has a situating sentence."""
    ctx = getattr(chunk, "context", None)
    text = chunk.text  # type: ignore[attr-defined]
    return f"{ctx}\n\n{text}" if ctx else text

build_chunk_rows 

build_chunk_rows(
    project_id: str, file_id: str, chunks: list
) -> List[ProjectChunk]

ORM rows for a file's chunks; ids match the Chroma ids ({file_id}_{index}).

Source code in adapta/services/rag.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def build_chunk_rows(project_id: str, file_id: str, chunks: list) -> List[ProjectChunk]:
    """ORM rows for a file's chunks; ids match the Chroma ids (``{file_id}_{index}``)."""
    return [
        ProjectChunk(
            id=f"{file_id}_{c.index}",
            file_id=file_id,
            project_id=project_id,
            chunk_index=c.index,
            text=c.text,
            context=getattr(c, "context", None),
            content_hash=content_hash(c.text),
            page=getattr(c, "page", None),
            section=getattr(c, "section", None),
            ocr=bool(getattr(c, "ocr", False)),
            ocr_confidence=getattr(c, "ocr_confidence", None),
            needs_review=_needs_review(c),
        )
        for c in chunks
    ]

persist_chunks async 

persist_chunks(
    db: "AsyncSession",
    project_id: str,
    file_id: str,
    chunks: list,
) -> int

Insert a file's chunk rows in db (flushed, not committed); returns the count.

Call this before the Chroma upsert: if the upsert fails the caller's rollback discards the rows, so Postgres never records chunks that have no vectors.

Source code in adapta/services/rag.py 
158
159
160
161
162
163
164
165
166
167
168
169
async def persist_chunks(
    db: "AsyncSession", project_id: str, file_id: str, chunks: list
) -> int:
    """Insert a file's chunk rows in ``db`` (flushed, not committed); returns the count.

    Call this *before* the Chroma upsert: if the upsert fails the caller's rollback
    discards the rows, so Postgres never records chunks that have no vectors.
    """
    rows = build_chunk_rows(project_id, file_id, chunks)
    db.add_all(rows)
    await db.flush()
    return len(rows)

count_project_chunks async 

count_project_chunks(
    db: "AsyncSession", project_id: str
) -> int

Chunk count for a project, read from Postgres (the source of truth).

Source code in adapta/services/rag.py 
172
173
174
175
176
177
178
179
180
async def count_project_chunks(db: "AsyncSession", project_id: str) -> int:
    """Chunk count for a project, read from Postgres (the source of truth)."""
    from sqlalchemy import func, select

    return (
        await db.scalar(select(func.count()).select_from(ProjectChunk).where(
            ProjectChunk.project_id == project_id
        ))
    ) or 0

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 
252
253
254
255
256
257
258
259
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 
828
829
830
831
832
833
834
835
836
837
838
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 
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
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: Optional[SentenceTransformer] = None
        self._query_prefix, self._passage_prefix = embedding_prefixes(model_name)

    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], *, kind: str = "passage") -> List[List[float]]:
        """Embed a batch of strings.  Returns one float vector per input.

        ``kind`` is ``"passage"`` (documents being indexed; the default) or ``"query"``;
        it only matters for models that use prefixes (see ``embedding_prefixes``).
        """
        if not texts:
            return []
        self._load()
        assert self._model is not None
        prefix = self._query_prefix if kind == "query" else self._passage_prefix
        try:
            if prefix:
                texts = [prefix + t for t in texts]
            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]``.

        This is the QUERY side (retrieval); indexing uses ``embed`` (passage side).
        Use ``embed`` directly for batches to avoid per-call overhead.
        """
        return self.embed([text], kind="query")[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], *, kind: str = "passage"
) -> List[List[float]]

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

kind is "passage" (documents being indexed; the default) or "query"; it only matters for models that use prefixes (see embedding_prefixes).

Source code in adapta/services/embeddings.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def embed(self, texts: List[str], *, kind: str = "passage") -> List[List[float]]:
    """Embed a batch of strings.  Returns one float vector per input.

    ``kind`` is ``"passage"`` (documents being indexed; the default) or ``"query"``;
    it only matters for models that use prefixes (see ``embedding_prefixes``).
    """
    if not texts:
        return []
    self._load()
    assert self._model is not None
    prefix = self._query_prefix if kind == "query" else self._passage_prefix
    try:
        if prefix:
            texts = [prefix + t for t in texts]
        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].

This is the QUERY side (retrieval); indexing uses embed (passage side). Use embed directly for batches to avoid per-call overhead.

Source code in adapta/services/embeddings.py 
90
91
92
93
94
95
96
def embed_one(self, text: str) -> List[float]:
    """Embed a single string.  Equivalent to ``embed([text])[0]``.

    This is the QUERY side (retrieval); indexing uses ``embed`` (passage side).
    Use ``embed`` directly for batches to avoid per-call overhead.
    """
    return self.embed([text], kind="query")[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 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
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: Optional[CrossEncoder] = 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 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
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

embedding_prefixes 

embedding_prefixes(model_name: str) -> tuple[str, str]

(query_prefix, passage_prefix) the model was trained with (E3.6).

E5 models are asymmetric: they expect "query: " / "passage: " in front of the text and degrade noticeably without them. Everything else takes the text as-is.

Source code in adapta/services/embeddings.py 
26
27
28
29
30
31
32
33
34
35
def embedding_prefixes(model_name: str) -> tuple[str, str]:
    """``(query_prefix, passage_prefix)`` the model was trained with (E3.6).

    E5 models are asymmetric: they expect ``"query: "`` / ``"passage: "`` in front of the
    text and degrade noticeably without them. Everything else takes the text as-is.
    """
    base = model_name.rsplit("/", 1)[-1].lower()
    if re.search(r"(^|[-_])e5([-_]|$)", base):
        return "query: ", "passage: "
    return "", ""

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 
106
107
108
109
110
111
112
113
114
115
@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 
169
170
171
172
173
174
175
176
177
178
179
180
@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)

Datasets (eval-set validation)

SFT dataset-row validation.

A dataset is a .jsonl file with one SFT row per line (prompt + response, optional system / metadata), checked against specs/schemas/eval_dataset.schema.json.

check_task 

check_task(task: str) -> str

Return task if it is a known eval task, else raise a typed 400.

Source code in adapta/services/datasets.py 
32
33
34
35
36
def check_task(task: str) -> str:
    """Return ``task`` if it is a known eval task, else raise a typed 400."""
    if task not in TASKS:
        raise InvalidRequest(message=f"task must be one of: {', '.join(TASKS)}.")
    return task

check_eval_set_size 

check_eval_set_size(path: Path) -> int

Count the non-blank rows of an uploaded eval set; raise a typed 400 outside eval_min_rows..eval_max_rows. Returns the count.

Source code in adapta/services/datasets.py 
39
40
41
42
43
44
45
46
47
48
49
def check_eval_set_size(path: Path) -> int:
    """Count the non-blank rows of an uploaded eval set; raise a typed 400 outside
    ``eval_min_rows``..``eval_max_rows``. Returns the count."""
    with path.open(encoding="utf-8") as f:
        n = sum(1 for line in f if line.strip())
    lo, hi = settings.eval_min_rows, settings.eval_max_rows
    if not lo <= n <= hi:
        raise InvalidRequest(
            message=f"An eval set needs {lo}-{hi} rows; this file has {n}.",
        )
    return n

validate_dataset 

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

Validate a JSONL file of SFT rows.

Returns (is_valid, error_message, num_samples).

Validation does not stop at the first bad row: it scans the file and reports up to :data:MAX_REPORTED_ERRORS problems at once, so an operator sees every issue in one pass instead of discovering them one re-upload at a time. With task="extract" every response must also be a JSON object (the reference the extraction metric compares keys against).

Source code in adapta/services/datasets.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def validate_dataset(path: Path, task: Optional[str] = None) -> tuple[bool, Optional[str], int]:
    """Validate a JSONL file of SFT rows.

    Returns (is_valid, error_message, num_samples).

    Validation does **not** stop at the first bad row: it scans the file and reports
    up to :data:`MAX_REPORTED_ERRORS` problems at once, so an operator sees every
    issue in one pass instead of discovering them one re-upload at a time. With
    ``task="extract"`` every ``response`` must also be a JSON object (the reference
    the extraction metric compares keys against).
    """
    try:
        schema = _load_schema()
        num_samples = 0
        errors: list[str] = []
        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 = _validate_row(i, obj, schema, task)
                if row_errors:
                    errors.extend(row_errors)
                    if len(errors) >= MAX_REPORTED_ERRORS:
                        break
                    continue
                num_samples += 1

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

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

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 is NOT hardcoded here: prompts are rendered by the named template carried on the request (InferenceRequest.chat_template), resolved from the model's catalog entry. The engine is a pure execution wrapper; which template a model uses — and how to onboard a non-ChatML family — lives in adapta/core/chat_templates.py.

Message dataclass

Single chat turn.

role must be one of the OpenAI roles (user, assistant, system) — the active chat template embeds it literally into the turn wrapper, so an unrecognized role produces malformed output silently.

Source code in adapta/core/inference.py 
47
48
49
50
51
52
53
54
55
56
57
@dataclass
class Message:
    """Single chat turn.

    ``role`` must be one of the OpenAI roles (``user``, ``assistant``,
    ``system``) — the active chat template embeds it literally into the turn
    wrapper, so 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 named chat_template before being passed to llama-cpp. chat_template is a key into adapta.core.chat_templates (default ChatML); the serving backend sets it from the model's catalog entry so the prompt matches what the base was trained on.

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 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@dataclass
class InferenceRequest:
    """Parameters for one inference call.

    Messages are formatted via ``_format_chat_prompt`` using the named
    ``chat_template`` before being passed to llama-cpp.  ``chat_template`` is a
    key into ``adapta.core.chat_templates`` (default ChatML); the serving
    backend sets it from the model's catalog entry so the prompt matches what
    the base was trained on.

    ``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
    chat_template: str = chat_templates.DEFAULT_TEMPLATE
    # OpenAI-shape ``response_format`` dict (None = unconstrained) and the
    # llama-cpp grammar the serving backend compiles from it (JSON mode).
    response_format: Optional[dict] = None
    grammar: Optional[Any] = 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 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@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 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
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:
        """Render the request into the prompt string for its ``chat_template``.

        The engine holds no template literals — it delegates to the named
        template in ``chat_templates`` (default ChatML). Onboarding a model with
        a different format is a change there + a catalog entry, never here.
        """
        return chat_templates.render(
            request.chat_template, request.system_prompt, request.messages
        )

    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 chat_templates.default_stops(request.chat_template)
        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,
                grammar=request.grammar,
                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,
        response_format: Optional[dict] = 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,
                response_format=response_format,
            )

        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 chat_templates.default_stops(request.chat_template)
        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,
                grammar=request.grammar,
                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,
        chat_template: str = chat_templates.DEFAULT_TEMPLATE,
    ) -> 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.  ``chat_template``
        must match the one generation will use, or the count omits that family's
        turn-wrapper tokens; callers resolve it from the model's catalog entry."""
        req = InferenceRequest(
            messages=messages,
            model_name="",
            system_prompt=system_prompt,
            chat_template=chat_template,
        )
        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 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
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 chat_templates.default_stops(request.chat_template)
    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,
            grammar=request.grammar,
            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,
    response_format: Optional[dict] = 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 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
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,
    response_format: Optional[dict] = 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,
            response_format=response_format,
        )

    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 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
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 chat_templates.default_stops(request.chat_template)
    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,
            grammar=request.grammar,
            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 
338
339
340
341
342
343
344
345
346
347
348
349
350
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,
    chat_template: str = chat_templates.DEFAULT_TEMPLATE,
) -> 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. chat_template must match the one generation will use, or the count omits that family's turn-wrapper tokens; callers resolve it from the model's catalog entry.

Source code in adapta/core/inference.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
def count_prompt_tokens(
    self,
    model: Llama,
    messages: List[Message],
    system_prompt: Optional[str] = None,
    chat_template: str = chat_templates.DEFAULT_TEMPLATE,
) -> 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.  ``chat_template``
    must match the one generation will use, or the count omits that family's
    turn-wrapper tokens; callers resolve it from the model's catalog entry."""
    req = InferenceRequest(
        messages=messages,
        model_name="",
        system_prompt=system_prompt,
        chat_template=chat_template,
    )
    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 
373
374
375
376
377
378
379
380
381
382
383
384
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 grouping of a catalog model — a cosmetic label the console uses to bucket the base-model list (chat / code / reasoning).

This is orthogonal to a model's modality (text vs vision), which is the functional axis: modality drives the serving path, dataset-match check and training path, whereas ModelType only affects presentation. A vision base is model_type=CHAT with modality="vision".

Source code in adapta/core/model_manager.py 
29
30
31
32
33
34
35
36
37
38
39
40
class ModelType(str, Enum):
    """Capability *grouping* of a catalog model — a cosmetic label the console
    uses to bucket the base-model list (chat / code / reasoning).

    This is orthogonal to a model's ``modality`` (text vs vision), which is the
    *functional* axis: modality drives the serving path, dataset-match check and
    training path, whereas ``ModelType`` only affects presentation. A vision base
    is ``model_type=CHAT`` with ``modality="vision"``."""

    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 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@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 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
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] = {}
        # Estimated resident MiB per cache key (F1.6); only keys still in `_models` count.
        self._costs: Dict[str, int] = {}
        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 a serving entry for every catalog model.

        The base-model catalog (``model_catalog``) is the single source of truth
        for the model list, GGUF paths, mmproj, type and context window (A3.3);
        this derives the serving ``ModelConfig`` from it rather than re-declaring
        each model, so the two can never drift. Imported lazily to avoid the
        catalog↔model_manager cycle (the catalog imports ``ModelType`` from here).

        All entries are lazy-loaded; none are pre-warmed here.  ``preload_default_models``
        optionally warms the configured default model at startup.
        """
        from adapta.core.model_catalog import all_entries

        for entry in all_entries():
            self._configs[entry.name] = ModelConfig(
                name=entry.name,
                model_type=entry.model_type,
                path=entry.gguf_path(),
                context_length=entry.context_length,
                n_threads=settings.n_threads,
                n_gpu_layers=settings.n_gpu_layers,
                description=entry.notes,
                mmproj_path=entry.mmproj_path(),
            )

    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 _make_room_for(
        self, cache_key: str, model_path: Path, n_ctx: int, mmproj_path: Optional[Path]
    ) -> None:
        """Evict LRU models until a NEW model fits the memory budget, BEFORE loading it (F1.6).

        The count cap in ``_evict_lru_if_needed`` ran only after ``Llama(...)`` had already
        allocated, so peak memory was old + new. Raises ``ModelTooLargeForMemory`` when the
        model cannot fit even alone. No budget (no cgroup limit, setting 0) = count cap only."""
        budget = memory_budget.budget_mb()
        cost = (
            memory_budget.estimate_cost_mb(model_path, n_ctx, mmproj_path)
            if budget is not None
            else 0
        )
        self._costs[cache_key] = cost
        if budget is not None and cost > budget:
            raise ModelTooLargeForMemory(
                message=(
                    f"Model needs about {cost} MiB but the memory budget for loaded models is "
                    f"{budget} MiB. Use a smaller base, or raise the app memory limit "
                    "(or ADAPTA_MODEL_MEMORY_BUDGET_MB)."
                ),
                internal_detail=f"model={cache_key} cost_mb={cost} budget_mb={budget}",
            )
        if cache_key not in self._models:
            self._evict_lru_if_needed()  # count cap, now before the allocation
        while budget is not None and self._models:
            used = sum(c for k, c in self._costs.items() if k in self._models and k != cache_key)
            if used + cost <= budget:
                return
            old_key, _ = self._models.popitem(last=False)
            if old_key == cache_key:  # force_reload of the LRU entry: it is being replaced
                continue
            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 (memory budget %d MiB, need %d MiB)", old_key, budget, cost
            )

    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
                    )
                self._make_room_for(cache_key, model_path, n_ctx, config.mmproj_path)
                loop = asyncio.get_event_loop()
                # Hand the Llama back through a list, NOT the executor Future's result: with the
                # real Llama the Future stays referenced after the await (measured, F1.6), so an
                # evicted model was never freed -- ~0.7 GB anon + the weights leaked per load.
                holder: list = []
                await loop.run_in_executor(None, lambda: holder.append(Llama(**llama_kwargs)))
                model = holder.pop()

                # 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

    async def evict(self, model_name: str, adapter_path: Optional[str] = None) -> bool:
        """Drop one (base, adapter) variant from the cache (E4.5). Wraps ``unload_model`` with
        the same cache key ``load_model`` uses; a request already holding the Llama finishes,
        the next one reloads. Returns whether the variant was loaded."""
        cache_key = self._cache_key(self._resolve_serving_name(model_name), adapter_path)
        async with self._load_lock:
            loaded = cache_key in self._models
            await self.unload_model(cache_key)
        return loaded

    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)

    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 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
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
                )
            self._make_room_for(cache_key, model_path, n_ctx, config.mmproj_path)
            loop = asyncio.get_event_loop()
            # Hand the Llama back through a list, NOT the executor Future's result: with the
            # real Llama the Future stays referenced after the await (measured, F1.6), so an
            # evicted model was never freed -- ~0.7 GB anon + the weights leaked per load.
            holder: list = []
            await loop.run_in_executor(None, lambda: holder.append(Llama(**llama_kwargs)))
            model = holder.pop()

            # 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 
366
367
368
369
370
371
372
373
374
375
376
377
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
evict async 
evict(
    model_name: str, adapter_path: Optional[str] = None
) -> bool

Drop one (base, adapter) variant from the cache (E4.5). Wraps unload_model with the same cache key load_model uses; a request already holding the Llama finishes, the next one reloads. Returns whether the variant was loaded.

Source code in adapta/core/model_manager.py 
379
380
381
382
383
384
385
386
387
async def evict(self, model_name: str, adapter_path: Optional[str] = None) -> bool:
    """Drop one (base, adapter) variant from the cache (E4.5). Wraps ``unload_model`` with
    the same cache key ``load_model`` uses; a request already holding the Llama finishes,
    the next one reloads. Returns whether the variant was loaded."""
    cache_key = self._cache_key(self._resolve_serving_name(model_name), adapter_path)
    async with self._load_lock:
        loaded = cache_key in self._models
        await self.unload_model(cache_key)
    return loaded
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 
389
390
391
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 
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
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 
420
421
422
def get_model_config(self, model_name: str) -> Optional[ModelConfig]:
    """Get model configuration"""
    return self._configs.get(model_name)
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 
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
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 
441
442
443
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 
445
446
447
448
449
450
451
452
453
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}")