From 25d8e3b7e5df43ffc2fdcc6da6af16dd7bcd137f Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 5 Jul 2026 23:30:36 +0000 Subject: [PATCH 1/2] 022: registry-driven LLM serving + operator model selection (T461-T488, offline slice) Make the text-generation engine registry-driven like the other four: promoting an LLM version from the console IS the go-live action (FR-254/255), with honest served identity end-to-end and base+LoRA-adapter resolution from lineage. Setup + foundational: - platformlib/store: ActiveServingLLM singleton pointer table + accessors (T461; mirrored in infra/postgres/init.sql; additive, no version bump) - scripts/register_base_gguf.py: register local zoo bases as kind=full-model versions with base_id tags, idempotent, never downloads (T462, R2/R7) - platformlib/llmresolve.py: ONE registry walk (name -> @serving -> kind -> base) shared by agent + gateway so the two resolution sites cannot drift - hostagent/serving_llm.py: cold-load resolver -> {base_gguf, adapter_gguf, model_name, version}; ResolutionError (refuse, FR-265) vs ResolutionUnavailable (env fallback, surfaced) (T463) US1 promote -> serve: - llama adapter rebinds base/adapter/alias/version from the registry at each cold load (TTL-cached on the health path); MODEL/LORA/MODEL_ALIAS env knobs become the fallback default; spawn appends --lora for adapters (T465) - swap.reload_serving_llm: cross-tenant preempt_for reuse + the same-tenant FORCE-RELOAD (unload -> ensure_loaded) preempt_for silently no-ops on; idempotent no-op, target-probe before evict, jobs never preempted, operator confirm (preempt=true) for any displacement (T466, PR #64 s1) - agent POST /control/reload (secret-gated, both transports) (T466) - gateway promote route: pre-gate resolvability refusal (409, alias unmoved), pointer write + reload request after a promoted alias move (T467, T474) US2 honest identity: - agent /engines/llm/health reports model_name + registry_version + base + adapter of what is ACTUALLY resident (T469) - gateway gpu_state / serving/state / /infer response / prediction logging all consume the agent-reported identity; degrade to unknown, never a config guess (T470/T471 - kills the live mislabeled-window divergence) US3/US4 fine-tunes first-class: - adapter path: base_model lineage -> registered full-model base, single hop, adapter chains refused (T473); finetune flow stamps task/serving_engine at registration (T476); scripts/backfill_llm_task_tags.py for legacy versions, idempotent + non-clobber (T477); kind->task fallback in resolve/list (T478); serving/tasks carries kind+lineage and filters text-generation to the active pointer - no stale duplicate LLM entries (T479, FR-276) US5 console switch (offline slice): - PromoteGate: promote-as-switch with the ConfirmDialog naming the displaced model, resident-vs-promoted delta, reload outcome surfaced (T482) - allow-list delta EMPTY: rides existing promote + serving/state (T483) - StreamPanel shows base-vs-adapter provenance (FR-268/274) Polish: - boundary guards: FR-273 no new resident process; FR-275 the policy/auto- promote path structurally cannot switch the served LLM (T485) - README serving section + tasks.md; AGENT_CONTROL_SECRET into gateway env Tests: 534 passed, 39 skipped (was 471/39); the 3 remaining failures are the pre-existing PIL-missing environment ones, unchanged. next build green. [HW] tail (T484/T489) + the live quickstart drills stay open for the GPU box. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013hoYEa8FN5s8eGqa78Fxjx --- README.md | 23 ++ docker-compose.yml | 4 + gateway/app/registry.py | 146 +++++++++- gateway/app/routers/infer.py | 74 +++--- gateway/app/routers/models.py | 34 ++- gateway/app/routers/stream.py | 22 +- gateway/app/serving.py | 76 +++++- gateway/app/settings.py | 6 +- hostagent/adapters/llama.py | 191 +++++++++++--- hostagent/main.py | 19 ++ hostagent/serving_llm.py | 134 ++++++++++ hostagent/swap.py | 70 +++++ infra/postgres/init.sql | 8 + platformlib/llmresolve.py | 141 ++++++++++ platformlib/store.py | 52 +++- scripts/backfill_llm_task_tags.py | 74 ++++++ scripts/register_base_gguf.py | 83 ++++++ .../022-registry-driven-llm-serving/tasks.md | 53 ++-- tests/_llmregistry.py | 132 ++++++++++ tests/test_backfill_llm_tags.py | 153 +++++++++++ tests/test_llama_registry_binding.py | 187 +++++++++++++ tests/test_llm_serving_boundaries.py | 99 +++++++ tests/test_serving_llm_gateway.py | 249 ++++++++++++++++++ tests/test_serving_llm_reload.py | 153 +++++++++++ tests/test_serving_llm_resolver.py | 193 ++++++++++++++ training/flows/finetune.py | 5 + ui/components/models/PromoteGate.tsx | 145 +++++++++- ui/components/serving/StreamPanel.tsx | 7 + ui/components/serving/types.ts | 11 +- 29 files changed, 2423 insertions(+), 121 deletions(-) create mode 100644 hostagent/serving_llm.py create mode 100644 platformlib/llmresolve.py create mode 100644 scripts/backfill_llm_task_tags.py create mode 100644 scripts/register_base_gguf.py create mode 100644 tests/_llmregistry.py create mode 100644 tests/test_backfill_llm_tags.py create mode 100644 tests/test_llama_registry_binding.py create mode 100644 tests/test_llm_serving_boundaries.py create mode 100644 tests/test_serving_llm_gateway.py create mode 100644 tests/test_serving_llm_reload.py create mode 100644 tests/test_serving_llm_resolver.py diff --git a/README.md b/README.md index 1d7c343..893dede 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,9 @@ a heavier reference platform, sized to a laptop. > **020 stack-remediation** landed on top: the archived-upstream object store exited to **Garage** > (migrated + decommissioned), the serving framework left the children ("Bento-ectomy", byte-identical > golden gates), and the agent's HTTP runtime was measured on hardware (verdict: keep-stdlib). +> **021** added the loop-native operator console; **022** makes the LLM **registry-driven** like the +> other engines (promote → serve, base + LoRA-adapter resolution, honest served identity — offline +> slice; the on-hardware switch drills are the [HW] tail). > Constitution `v1.5.1`. Reference stack on MLflow `3.14.0`. > > Per-increment detail lives under [`specs/`](specs/) (each has `spec.md` / `plan.md` / `tasks.md`); @@ -118,6 +121,26 @@ with the same serving tags **plus lineage** (`base_model`, parent run) so adapte Promotion to `@serving` is **operator-driven** (alias-based) and, since **011**, runs through an **evaluation gate** (below). +**The LLM is registry-driven too (022).** Pre-022 the `llama.cpp` engine was the sole outlier — it +loaded a fixed GGUF from the `MODEL` env and ignored the registry, so promoting an LLM did nothing. +Now **promote = go live**: promoting a text-generation version from the console moves its +`@serving` alias, records it as the **active serving LLM** (a one-row platform pointer in the +relational store — it spans model *names*, base vs fine-tune), and triggers an **immediate +controlled reload** on the agent (evict → load under admission; displacing a resident serving +model is operator-confirmed; a running job is never preempted — the switch defers instead). At +each cold load the adapter resolves the active model's `@serving` version from the registry: +a `kind=full-model` version serves `-m `; a `kind=lora-adapter` fine-tune serves +`-m --lora `, with the **base resolved automatically from `base_model` lineage** +to a registered base GGUF (`scripts/register_base_gguf.py` registers the local zoo bases; +an unresolvable/absent base **refuses the promotion** and leaves the served LLM unchanged — +nothing is auto-downloaded). Served identity is **honest end-to-end**: the agent reports the +actually-loaded model + registry version in its health, and `/serving/state`, the `/infer` +response, and every logged prediction consume that agent-reported identity (monitoring windows key +on the model that really produced each output). The `MODEL`/`LORA`/`MODEL_ALIAS` env knobs remain +only as the **fallback default** for when no serving LLM has been selected. Legacy pre-022 LLM +versions are recognized by artifact kind and can be batch-tagged with +`scripts/backfill_llm_task_tags.py` (idempotent, never clobbers an existing tag). + ## Run it See [quickstart](specs/001-mlops-platform/quickstart.md). diff --git a/docker-compose.yml b/docker-compose.yml index 1a41f06..36638ac 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -129,6 +129,10 @@ services: # since host.docker.internal can't cross WSL distros. AGENT_URL: ${AGENT_URL:-http://host.docker.internal:8100} SERVING_MODEL: ${SERVING_MODEL:-qwen2.5-7b-instruct-q4_k_m} + # 022 (FR-255): the gated promote requests the agent's /control/reload so a promoted LLM + # goes live at once. Empty when the agent runs open (the default); must MATCH the agent's + # AGENT_CONTROL_SECRET when one is set (sent as X-Agent-Control). + AGENT_CONTROL_SECRET: ${AGENT_CONTROL_SECRET:-} # Garage/S3 for the dataset registry (US3) — the store-minted pair (T406 cutover). MLFLOW_S3_ENDPOINT_URL: http://garage:3900 AWS_ACCESS_KEY_ID: "${GARAGE_ACCESS_KEY_ID:?set GARAGE_ACCESS_KEY_ID in .env (scripts/gen_secrets --record-garage)}" diff --git a/gateway/app/registry.py b/gateway/app/registry.py index ef9d7d7..e3c5888 100644 --- a/gateway/app/registry.py +++ b/gateway/app/registry.py @@ -8,14 +8,21 @@ The MLflow server version is pinned in infra/mlflow (3.14.0, 007 US1); this client matches it. """ import os +import time from typing import Optional from mlflow.exceptions import MlflowException from mlflow.tracking import MlflowClient +from platformlib import llmresolve + MLFLOW_TRACKING_URI = os.getenv("MLFLOW_TRACKING_URI", "http://mlflow:5000") SERVING_ALIAS = "serving" +# 022: the configured default base — what serves when no serving-LLM was ever selected (the +# ActiveServingLLM pointer's documented unset behavior, data-model.md). +DEFAULT_LLM = os.getenv("SERVING_MODEL", "qwen2.5-7b-instruct-q4_k_m") + # 009 US1 — registry routing metadata (FR-074). Every version SHOULD carry these version tags; the # gateway resolves "the serving model+engine for task X" from them rather than a hard-coded # endpoint→model map (FR-075). Versions registered before 009 lack them and are tolerated (task=None). @@ -128,6 +135,91 @@ def promote(name: str, version: str, override: bool = False) -> dict: "promoted": True, "verdict": verdict} +# --- 022 US1/US4: the active serving-LLM pointer + LLM target resolution --------------------------- +# +# The pointer (data-model.md §ActiveServingLLM) lives in the relational store (T461), NOT in an +# MLflow alias — it spans model NAMES (base vs fine-tune), which a per-model alias cannot express +# (research R1). Reads are best-effort (unset/unreachable ⇒ the configured default base — the +# documented unset behavior); the WRITE fails loud (RegistryError) so a promote whose go-live +# pointer didn't stick is surfaced, never silent. + + +def _store(): + from platformlib import store + return store + + +def get_serving_llm() -> Optional[dict]: + """The active serving-LLM selection {model_name, selected_at, selected_by}, or None when unset + (or the store is unreachable — both read as 'the configured default base serves').""" + try: + s = _store() + conn = s.connect() + try: + return s.get_serving_llm(conn) + finally: + conn.close() + except Exception: # noqa: BLE001 — best-effort read; unset ⇒ default base + return None + + +def set_serving_llm(model_name: str, actor: str = "operator") -> dict: + """Record `model_name` as the active serving LLM (022 US1 — the gated promote IS the go-live + action; this is its persistence half, the agent reload is the other). Fail-loud.""" + try: + s = _store() + conn = s.connect() + try: + s.set_serving_llm(conn, model_name, time.time(), actor) + finally: + conn.close() + except Exception as e: # noqa: BLE001 — normalize driver/connection errors + raise RegistryError(f"active serving-LLM pointer write failed: {e}") from e + return {"model_name": model_name, "selected_by": actor} + + +def active_serving_llm_name() -> str: + """The model name that SHOULD be serving text-generation: the pointer, else the configured + default base (FR-276 — the disambiguator when several LLM models hold a promoted alias).""" + rec = get_serving_llm() + return rec["model_name"] if rec else DEFAULT_LLM + + +def llm_target_info(name: str, version: str) -> Optional[dict]: + """Pre-promotion resolution check for a text-generation version (022 T467/T474, FR-265). + + Returns None when (name, version) is not a text-generation target (task tag or inferred kind + — T478), else {task, kind, base, error}: `error` is set when the target cannot resolve at the + REGISTRY level (adapter with no/unresolvable base — the artifact-presence half is the agent's + probe). The promote route refuses on `error` BEFORE the gate runs, so the alias — and the + currently-served LLM — stay unchanged.""" + c = _client() + try: + mv = c.get_model_version(name, str(version)) + except MlflowException as e: + raise RegistryError(str(e)) from e + tags = dict(mv.tags or {}) + task = tags.get(TASK_TAG) or llmresolve.task_from_kind(tags) + if task != llmresolve.TEXT_GENERATION: + return None + kind = llmresolve.effective_kind(tags) + info = {"task": task, "kind": kind, "base": None, "error": None} + if kind == llmresolve.KIND_ADAPTER: + base_ref = tags.get(llmresolve.BASE_MODEL_TAG) + if not base_ref: + info["error"] = (f"{name} v{version} is a LoRA adapter with no base_model lineage — " + f"its base cannot be resolved (FR-264/265)") + return info + try: + bmv = llmresolve.resolve_base_version(c, base_ref) + info["base"] = {"name": bmv.name, "version": str(bmv.version), "source": bmv.source} + except llmresolve.LLMResolutionError as e: + info["error"] = str(e) + except MlflowException as e: + raise RegistryError(str(e)) from e + return info + + # --- 009 US1: task/serving-engine routing metadata ------------------------------------------------ def set_version_tags(name: str, version: str, tags: dict) -> dict: @@ -218,15 +310,50 @@ def resolve_serving_target(task: str, prefer_name: Optional[str] = None) -> Opti return entry # the configured backend model wins over any other task-mate if first is None: first = entry + if first is None and task == llmresolve.TEXT_GENERATION: + # 022 T478 (FR-267) defense-in-depth: a legacy promoted LLM version with NO task tag + # (kind/format only, pre-backfill) is invisible to the tag search above — infer its task + # from the artifact kind so no valid LLM version is stranded as unroutable. + try: + models = c.search_registered_models() + except MlflowException as e: + raise RegistryError(str(e)) from e + for m in models: + sv = (dict(m.aliases or {})).get(SERVING_ALIAS) + if not sv: + continue + try: + mv = c.get_model_version(m.name, sv) + except MlflowException: + continue + tags = dict(mv.tags or {}) + if tags.get(TASK_TAG) or llmresolve.task_from_kind(tags) != task: + continue # tagged versions were already considered by the search above + entry = {"name": m.name, "version": str(sv), "task": task, + "serving_engine": _resolve_engine(c, mv), "source": mv.source} + if prefer_name is not None and m.name == prefer_name: + return entry + if first is None: + first = entry return first +#: The lineage tags surfaced per LLM serving target (FR-268: base / parent / dataset provenance). +_LINEAGE_KEYS = ("base_model", "dataset_name", "dataset_version", "parent_version", "lineage") + + def list_tasks() -> list: """Distinct serving tasks discovered from the registry — one entry per registered model's `@serving` version, reporting its `task`/`serving_engine` (FR-077). The Infer tab renders one panel per task; a serving version with no `task` tag (legacy, pre-009) is reported as task=None so the UI degrades it to a read-only "no renderer" placeholder rather than breaking the tab. - """ + + 022 (T478/T479): a text-generation version's task is INFERRED from its artifact kind when the + tag is missing (legacy fine-tunes render a real LLM panel, not "no renderer" — FR-267), each + LLM entry carries its base-vs-adapter `kind` + lineage (FR-268), and the text-generation rows + are filtered to the ACTIVE serving-LLM pointer (FR-276): a promote moves only its own model's + `@serving` alias, so several LLM models can each retain one — but only the active-pointer + model is the live target; a previously-active LLM must not linger as a stale duplicate.""" c = _client() try: models = c.search_registered_models() @@ -242,10 +369,21 @@ def list_tasks() -> list: except MlflowException: continue tags = dict(mv.tags or {}) - out.append({ + task = tags.get(TASK_TAG) or llmresolve.task_from_kind(tags) + entry = { "model": m.name, "version": str(sv), - "task": tags.get(TASK_TAG), + "task": task, "serving_engine": _resolve_engine(c, mv), - }) + } + if task == llmresolve.TEXT_GENERATION: + entry["kind"] = llmresolve.effective_kind(tags) + lineage = {k: tags[k] for k in _LINEAGE_KEYS if tags.get(k)} + entry["lineage"] = lineage or None + out.append(entry) + text_gen = [e for e in out if e["task"] == llmresolve.TEXT_GENERATION] + if len(text_gen) > 1: + active = active_serving_llm_name() + keep = next((e for e in text_gen if e["model"] == active), text_gen[0]) + out = [e for e in out if e["task"] != llmresolve.TEXT_GENERATION or e is keep] return out diff --git a/gateway/app/routers/infer.py b/gateway/app/routers/infer.py index 3477947..a31f933 100644 --- a/gateway/app/routers/infer.py +++ b/gateway/app/routers/infer.py @@ -12,12 +12,12 @@ from .. import quality, registry, tracing from ..serving import ( - SERVING_MODEL, ModelTooLargeError, ServingBusyError, ServingError, gpu_state, health, + llm_identity, run_inference, ) @@ -35,29 +35,24 @@ class InferRequest(BaseModel): preempt: bool = False # 017: opt-in swap — evict a resident *serving* model first (default 008 refuse) -async def _resolve_serving_version() -> str | None: - """Registry version currently serving the LLM (FR-006/FR-075). - - /infer always proxies to the single llama supervisor configured by SERVING_MODEL, so SERVING_MODEL's - `@serving` alias is authoritative for the reported version — resolve it FIRST. This preserves pre-009 - /infer behavior exactly (T156): the promoted version is reported even when it predates 009's `task` - tags, and a *different* promoted text-generation model can never be mis-reported as the served one. - (The earlier task-first resolve mis-fired when SERVING_MODEL's `@serving` version was untagged but a - text-generation tag existed elsewhere — `resolve_serving_target` then returned that other model's - version, since a candidate must be BOTH `task`-tagged AND its model's `@serving` version.) - - The task-based resolve remains only as a fallback for the edge where SERVING_MODEL has nothing - promoted to `@serving`. Best-effort — any failure yields None (the response just omits the version). - """ - try: - served = await run_in_threadpool(registry.get_serving, SERVING_MODEL) - if served: - return served["version"] - target = await run_in_threadpool( - registry.resolve_serving_target, "text-generation", SERVING_MODEL) - return target["version"] if target else None - except Exception: - return None +async def _served_identity() -> dict: + """The identity every surface attributes this inference to (022 US2, FR-260/261/262): the + AGENT-REPORTED {serving_model, serving_version} — the agent is the only component that knows + what is actually resident, so the response's registry_model/registry_version, the logged + prediction, and `/serving/state` all name the same model+version (the pre-022 divergence — + `model: ops-bot-v2` logged as `registry_model: qwen…` — cannot recur). When the agent doesn't + report a registry version (env-default binding on a legacy agent), the version falls back to + the reported MODEL's own `@serving` alias — still keyed to the served name, never to the + fixed SERVING_MODEL config.""" + ident = await llm_identity() + if ident["serving_model"] != "unknown" and ident["serving_version"] is None: + try: + served = await run_in_threadpool(registry.get_serving, ident["serving_model"]) + if served: + ident["serving_version"] = served["version"] + except Exception: + pass # best-effort — the response just omits the version + return ident @router.post("/infer") @@ -69,7 +64,7 @@ async def infer(req: InferRequest): """ start_ns = time.time_ns() result = None - registry_version = None + served_model, registry_version = "unknown", None # Pessimistic default: any UNEXPECTED failure (e.g. a malformed supervisor response) keeps the trace # errored — only the known-good path flips it to OK just before returning (006 Codex review). trace_status = "ERROR" @@ -106,12 +101,15 @@ async def infer(req: InferRequest): LOAD_LATENCY.observe(result["load_ms"] / 1000.0) INFER_LATENCY.observe(result.get("infer_ms", 0) / 1000.0) INFER_REQUESTS.labels(status="ok").inc() - registry_version = await _resolve_serving_version() + # 022 US2 (FR-260/261/262): attribute everything — response identity, prediction log, + # trace — to the AGENT-REPORTED served identity, not the fixed SERVING_MODEL config. + ident = await _served_identity() + served_model, registry_version = ident["serving_model"], ident["serving_version"] trace_status, outcome = "OK", "completed" # success is known — flip the pessimistic default # 013/FR-119: log the served prediction off the request path (fire-and-forget, fail-open) so it # can be scored later against a delayed label. Returns a synchronous id regardless of store state. prediction_id = quality.log_prediction( - SERVING_MODEL, registry_version, "text-generation", req.prompt, result.get("text")) + served_model, registry_version, "text-generation", req.prompt, result.get("text")) # 016 (FR-146): also route the prompt to the bounded recoverable-input capture (uniform replay # corpus across modalities), so it can be shadow-replayed. The served decoding settings ride along # so a replay reproduces them, not the scorer defaults. Fire-and-forget + fail-open. @@ -119,7 +117,7 @@ async def infer(req: InferRequest): options={"max_tokens": req.max_tokens, "temperature": req.temperature}) return { "status": "completed", - "registry_model": SERVING_MODEL, + "registry_model": served_model, "registry_version": registry_version, "prediction_id": prediction_id, **result, @@ -128,7 +126,7 @@ async def infer(req: InferRequest): attrs = { "max_tokens": req.max_tokens, "temperature": req.temperature, - "model": SERVING_MODEL, + "model": served_model, "status": outcome, } if result: @@ -156,10 +154,22 @@ async def serving_health(): @router.get("/serving/state") async def serving_state(): """GPU/lease state for the Infer status line (008 US3, FR-068): {holder, resident, - serving_model, serving_version}. holder ∈ {llm, vision, training, null}; the version is the - registry @serving pointer. Read-only; the API key never reaches the browser (BFF contract).""" + serving_model, serving_version, base, adapter}. holder ∈ {llm, vision, training, null}. + + 022 (FR-260, T470): the model+version are the AGENT-REPORTED served identity (what is actually + resident, or what the next load would serve when cold) — `unknown` when the agent is + unreachable, never a stale config guess. `base`/`adapter` expose a served fine-tune's resolved + provenance (FR-274). Read-only; the API key never reaches the browser (BFF contract).""" state = await gpu_state() - state["serving_version"] = await _resolve_serving_version() + if state["serving_model"] != "unknown" and state["serving_version"] is None: + # Legacy-agent tolerance: no registry_version reported → the served NAME's own @serving + # alias (still keyed to the reported name — the same rule as _served_identity). + try: + served = await run_in_threadpool(registry.get_serving, state["serving_model"]) + if served: + state["serving_version"] = served["version"] + except Exception: + pass return state diff --git a/gateway/app/routers/models.py b/gateway/app/routers/models.py index ac4a626..fb00cee 100644 --- a/gateway/app/routers/models.py +++ b/gateway/app/routers/models.py @@ -13,7 +13,7 @@ from prometheus_client import Counter from pydantic import BaseModel, Field -from .. import evaluation, quality, registry, shadow +from .. import evaluation, quality, registry, serving, shadow from ..settings import TRAINER_URL router = APIRouter() @@ -32,6 +32,10 @@ class RegisterRequest(BaseModel): class PromoteRequest(BaseModel): version: str override: bool = False # 011 FR-104: explicitly bypass a hard-gate block (promote-with-regression) + # 022 (FR-258): promoting a text-generation version IS the served-LLM switch; when it would + # displace a RESIDENT serving model the agent requires this operator confirmation (the console + # sets it after the ConfirmDialog naming the displaced model). Ignored for non-LLM versions. + preempt: bool = False class EvaluateRequest(BaseModel): @@ -84,19 +88,45 @@ def promote(name: str, req: PromoteRequest): """Promote a version to serving, **through the evaluation gate** (011 FR-105). The response carries the gate verdict and whether the alias moved (`promoted`): a default hard-gate block keeps the alias put with `promoted=false`; `pass`/`warn` (or an explicit `override`) moves it. US1 inference - then resolves to the serving version (FR-006).""" + then resolves to the serving version (FR-006). + + 022 (FR-255 — promote = go live, one operator action): for a TEXT-GENERATION version this route + is also the served-LLM switch. In order: (1) an adapter whose base does not resolve is refused + HERE, before the gate — the alias and the currently-served LLM stay unchanged (FR-265); + (2) a promoted alias move also writes the ActiveServingLLM pointer (T461/T464); (3) the agent + is asked for the immediate controlled reload — refused/deferred reloads (job holder, missing + operator confirm) are surfaced in `serving_llm`, never silent (FR-259). This route is the ONLY + go-live surface: the retraining policy path calls `registry.promote` directly and so can gate a + candidate but can never switch the served LLM (FR-275).""" try: versions = registry.list_versions(name) except registry.RegistryError as e: raise HTTPException(status_code=502, detail=f"registry error: {e}") if not any(v["version"] == str(req.version) for v in versions): raise HTTPException(status_code=404, detail=f"{name!r} has no version {req.version}") + try: + llm_target = registry.llm_target_info(name, req.version) + except registry.RegistryError as e: + raise HTTPException(status_code=502, detail=f"registry error: {e}") + if llm_target and llm_target.get("error"): + REGISTRY_OPS.labels(op="promote", status="refused").inc() + raise HTTPException(status_code=409, detail=f"promotion refused: {llm_target['error']}") try: res = registry.promote(name, req.version, override=req.override) except registry.RegistryError as e: REGISTRY_OPS.labels(op="promote", status="error").inc() raise HTTPException(status_code=502, detail=f"registry error: {e}") REGISTRY_OPS.labels(op="promote", status="blocked" if not res.get("promoted", True) else "ok").inc() + if llm_target and res.get("promoted"): + try: + registry.set_serving_llm(name, actor="operator") + res["serving_llm"] = {"active": name, "kind": llm_target["kind"], + "base": llm_target["base"], + "reload": serving.request_llm_reload(preempt=req.preempt)} + except registry.RegistryError as e: + # The alias moved but the go-live pointer didn't stick — surface it; a re-promote of + # the same version retries both halves idempotently. + res["serving_llm"] = {"active": None, "error": str(e)} return res diff --git a/gateway/app/routers/stream.py b/gateway/app/routers/stream.py index 62e1bba..f1e4a07 100644 --- a/gateway/app/routers/stream.py +++ b/gateway/app/routers/stream.py @@ -142,17 +142,23 @@ async def gen(): # The streamed tokens aren't buffered (the SSE bytes stay byte-identical), so the output is # left uncaptured — the prediction id + prompt + version are still logged for later labeling. if outcome == "completed": - # Fully off the response path: schedule the version-resolve + store write as a detached + # Fully off the response path: schedule the identity-resolve + store write as a detached # task so the generator's teardown (the chunked-encoding terminator) isn't delayed by a # registry call (Codex P2). The streamed bytes are already delivered + unaffected. async def _log(): - try: - served = await run_in_threadpool(registry.get_serving, serving.SERVING_MODEL) - version = served["version"] if served else None - except Exception: - version = None - quality.log_prediction(serving.SERVING_MODEL, version, "text-generation", - req.prompt, None) + # 022 US2 (FR-261): attribute the streamed prediction to the AGENT-REPORTED + # served identity — not the fixed SERVING_MODEL config (the divergence that + # corrupted a served fine-tune's quality window). Version falls back to the + # served NAME's own @serving alias when the agent reports none (legacy agent). + ident = await serving.llm_identity() + model, version = ident["serving_model"], ident["serving_version"] + if model != "unknown" and version is None: + try: + served = await run_in_threadpool(registry.get_serving, model) + version = served["version"] if served else None + except Exception: + version = None + quality.log_prediction(model, version, "text-generation", req.prompt, None) # 016 (FR-146): do NOT capture streamed prompts. The streamed output isn't logged # (prediction=None), so join_window excludes these as champion-unscorable — but a # capture would still consume the per-modality ring-buffer cap, evicting replayable diff --git a/gateway/app/serving.py b/gateway/app/serving.py index 681564c..5fb4e81 100644 --- a/gateway/app/serving.py +++ b/gateway/app/serving.py @@ -52,26 +52,88 @@ async def health() -> bool: return False +def _identity_from_health(h: dict) -> dict: + """The served-LLM identity fields from the agent's /engines/llm/health payload (022 US2, + contracts/agent-identity-and-allowlist.md). Pure so the honest-identity mapping unit-tests + without HTTP. The agent is the ONLY component that knows what is resident — nothing here + falls back to the fixed SERVING_MODEL config (that fallback WAS the live divergence bug).""" + return { + "serving_model": h.get("model_name") or h.get("model") or "unknown", + "serving_version": h.get("registry_version"), + "base": h.get("base"), + "adapter": h.get("adapter"), + } + + +_UNKNOWN_IDENTITY = {"serving_model": "unknown", "serving_version": None, + "base": None, "adapter": None} + + async def gpu_state() -> dict: """GPU state for the UI status line (008 FR-068): which tenant holds the single GPU slot, the - serving model name, and whether the LLM is resident. + serving model identity, and whether the LLM is resident. Sourced from the agent's per-engine `/health` (`lease_holder`, from `admission.holder()` since T364) — so one read reflects whichever tenant (llm, vision, asr, training) holds the GPU. - Key-free: the BFF contract is unchanged (no key reaches the browser). holder=None when unreadable. + 022 (FR-260): the serving model+version are the AGENT-REPORTED identity, degrading to + `unknown` when the agent is unreachable — never a stale config guess (T470). Key-free: the + BFF contract is unchanged (no key reaches the browser). holder=None when unreadable. """ - holder, resident, model = None, False, SERVING_MODEL + holder, resident, ident = None, False, dict(_UNKNOWN_IDENTITY) async with httpx.AsyncClient(timeout=5) as client: try: r = await client.get(f"{SERVING_URL}/health") - if r.status_code == 200: + if r.status_code in (200, 503): # 503 = engine unavailable, but the payload is honest h = r.json() resident = bool(h.get("resident")) holder = _HOLDER_LABEL.get(h.get("lease_holder"), h.get("lease_holder")) - model = h.get("model") or SERVING_MODEL - except httpx.HTTPError: + ident = _identity_from_health(h) + except (httpx.HTTPError, ValueError): + pass + return {"holder": holder, "resident": resident, **ident} + + +async def llm_identity() -> dict: + """The agent-reported served-LLM identity {serving_model, serving_version, base, adapter} — + what prediction logging attributes each served prediction to (022 US2, FR-261: the quality + window must key on the model+version that actually produced the output). `unknown`/None when + the agent is unreachable — a prediction is never logged under a config guess.""" + async with httpx.AsyncClient(timeout=5) as client: + try: + r = await client.get(f"{SERVING_URL}/health") + if r.status_code in (200, 503): + return _identity_from_health(r.json()) + except (httpx.HTTPError, ValueError): pass - return {"holder": holder, "resident": resident, "serving_model": model} + return dict(_UNKNOWN_IDENTITY) + + +def request_llm_reload(preempt: bool = False) -> dict: + """Ask the agent to make the newly selected serving-LLM live NOW (022 US1, FR-255 — the + gated promote is the go-live action; this is its reload half). Synchronous (the promote + handler runs in the threadpool). Never raises: the alias + pointer have already moved, so a + refused/failed reload is reported as status=deferred|unreachable with the agent's reason — + FR-259's refuse/defer vocabulary — for the operator to see and retry (re-promoting the same + version re-requests the reload idempotently).""" + headers = {} + if settings.AGENT_CONTROL_SECRET: + headers["X-Agent-Control"] = settings.AGENT_CONTROL_SECRET + try: + with httpx.Client(timeout=300) as client: + r = client.post(f"{settings.AGENT_URL}/control/reload", + json={"preempt": preempt}, headers=headers) + except httpx.HTTPError as e: + return {"status": "unreachable", + "reason": f"agent unreachable at {settings.AGENT_URL}: {e} — the switch is " + f"picked up on the next cold load"} + try: + body = r.json() + except ValueError: + body = {} + if r.status_code == 200: + return body # loaded | reloaded | swapped | noop (+ the served identity) + return {"status": "deferred", + "reason": body.get("error") or f"agent refused the reload ({r.status_code})"} async def run_inference(prompt: str, max_tokens: int = 256, temperature: float = 0.7, *, diff --git a/gateway/app/settings.py b/gateway/app/settings.py index f454614..0c19e4c 100644 --- a/gateway/app/settings.py +++ b/gateway/app/settings.py @@ -32,7 +32,11 @@ ASR_URL = f"{AGENT_URL}/engines/asr" TRAINER_URL = AGENT_URL # the jobs surface serves the legacy /train|/study|/batch|/shadow-replay aliases -import os as _os # noqa: E402 — the two non-topology settings +import os as _os # noqa: E402 — the non-topology settings SERVING_MODEL = _os.getenv("SERVING_MODEL", "qwen2.5-7b-instruct-q4_k_m") MLFLOW_TRACKING_URI = _os.getenv("MLFLOW_TRACKING_URI", "http://mlflow:5000") +# 022 (FR-255): the gated promote requests the agent's /control/reload so the newly selected +# serving-LLM goes live. Same secret vocabulary as the agent (sent as X-Agent-Control; the SWAP_ +# name survives as the deprecated pre-018 alias). +AGENT_CONTROL_SECRET = _os.getenv("AGENT_CONTROL_SECRET") or _os.getenv("SWAP_CONTROL_SECRET", "") diff --git a/hostagent/adapters/llama.py b/hostagent/adapters/llama.py index 4954cde..c4e0415 100644 --- a/hostagent/adapters/llama.py +++ b/hostagent/adapters/llama.py @@ -1,18 +1,25 @@ -"""LLM engine adapter (018 US2, T358) — llama.cpp, folded in from `serving/llama/supervisor.py`. +"""LLM engine adapter (018 US2, T358; 022 US1/US3, T465) — llama.cpp, registry-driven. -The load-on-demand → ready → drain → idle-release → reap lifecycle now lives ONCE in +The load-on-demand → ready → drain → idle-release → reap lifecycle lives ONCE in `hostagent.lifecycle`; this adapter supplies only the llama-specific bits: spawn a `llama-server` -on a **dynamic** port (no fixed :8081 to collide with a peer engine's child), probe its readiness, -and forward inference — REST and SSE — to it. Stdlib-only (the agent carries no pip deps). +on a **dynamic** port, probe its readiness, and forward inference — REST and SSE — to it. +Stdlib-only (the agent carries no pip deps; the resolver's mlflow/boto3 load lazily). -Byte-compatibility (FR-177): the forward request/response bodies and the SSE frames are identical to -the retired supervisor's, so the gateway's serving router + `/infer/stream` proxy change base URL -only. The SSE `done` frame in particular must keep the exact `"event": "done"` marker the gateway's -stream proxy scans for. +022 (FR-254/255): the served artifact is REGISTRY-RESOLVED at each cold load — `rebind()` asks +`hostagent.serving_llm` for the active serving-LLM and binds `self.model` (base GGUF), +`self.lora` (adapter GGUF or None) and the response identity (`self.alias` = the registry model +name, `self.registry_version`) before `spawn()` builds the command line. The `MODEL`/`LORA`/ +`MODEL_ALIAS` env knobs are now the FALLBACK DEFAULT only — they serve when the pointer is unset +(byte-for-byte today's behavior, SC-149) or when resolution infra is unreachable (surfaced in +health as `binding`, never silent). A pointer that names an UNRESOLVABLE target makes the engine +unavailable (fail loud, FR-265) — the swap target-probe then refuses rather than evicting a +working holder for a load that would fail. + +Byte-compatibility (FR-177/FR-271): the forward request/response bodies and the SSE frames keep +the retired supervisor's exact shapes; identity fields in /health are additive. Health source (T364): `admission` is the single in-process GPU authority, read ONLY to build the -byte-compatible `/engines/llm/health` GLOBAL holder + free-VRAM fields (the retired lockfile used -to be the shared cross-process source; with one owner, admission is the whole truth). +byte-compatible `/engines/llm/health` GLOBAL holder + free-VRAM fields. """ import json import os @@ -27,6 +34,11 @@ DEFAULT_MODEL = "~/models/gguf/Qwen2.5-7B-Instruct-Q4_K_M.gguf" DEFAULT_ALIAS = "qwen2.5-7b-instruct-q4_k_m" +#: How long a registry binding stays fresh (s). available() is on the hot /health path — the TTL +#: bounds pointer/registry reads to one per window; the promote-triggered reload forces a fresh +#: resolve regardless (rebind(force=True)), so a switch is never TTL-delayed. +REBIND_TTL_S = float(os.getenv("LLM_REBIND_TTL_S", "5")) + class LlamaAdapter: """Adapter for the text-generation (llm) engine. Implements the duck-typed @@ -39,48 +51,136 @@ class LlamaAdapter: verbs = ("infer",) # POST /engines/llm/infer stream_verbs = ("infer",) # POST /engines/llm/infer/stream (SSE) - def __init__(self, admission=None): + def __init__(self, admission=None, resolver=None): self.llama_bin = os.path.expanduser( os.getenv("LLAMA_BIN", "~/llama.cpp/build/bin/llama-server")) - self.model = os.path.expanduser(os.getenv("MODEL", DEFAULT_MODEL)) - self.alias = os.getenv("MODEL_ALIAS", DEFAULT_ALIAS) + # Env-configured DEFAULTS (022): what serves when no serving-LLM is selected. LORA is the + # generalized prototype knob — a default adapter, normally unset. + self._env_model = os.path.expanduser(os.getenv("MODEL", DEFAULT_MODEL)) + self._env_lora = os.path.expanduser(os.getenv("LORA")) if os.getenv("LORA") else None + self._env_alias = os.getenv("MODEL_ALIAS", DEFAULT_ALIAS) + # The LIVE binding (what the next spawn serves) — starts at the defaults; rebind() moves it. + self.model = self._env_model + self.lora = self._env_lora + self.alias = self._env_alias + self.registry_version = None self.ngl = os.getenv("NGL", "999") self.ctx = os.getenv("CTX", "4096") self.vram_budget_gb = vram_budget_gb() # FR-207: the single shared budget resolver self._admission = admission - self._port = None # set on spawn(); the child's dynamic port + self._resolver = resolver # injectable for tests; None → hostagent.serving_llm.resolve + self._bind_error = None # ResolutionError text → available() reports (False, reason) + self._bind_note = None # degradation note (resolution infra unreachable → env default) + self._bound_at = None # monotonic time of the last successful rebind (TTL) + self._loaded = None # identity captured at spawn() — what is ACTUALLY resident + self._port = None # set on spawn(); the child's dynamic port + + # -- 022 registry binding (T465) --------------------------------------------------------------- + def rebind(self, force: bool = False) -> None: + """Resolve the active serving-LLM and bind base/adapter/identity for the next spawn. + + TTL-cached (REBIND_TTL_S) because available() runs on every health poll; the reload verb + passes force=True so a promote is never stale. Never raises — a ResolutionError is recorded + in `_bind_error` (available() fails loud), an unreachable store/registry falls back to the + env defaults with `_bind_note` set (surfaced, honest: identity reports the actual binding). + """ + now = time.monotonic() + if not force and self._bound_at is not None and (now - self._bound_at) < REBIND_TTL_S: + return + resolver = self._resolver + if resolver is None: + from hostagent import serving_llm + resolver = serving_llm.resolve + from hostagent import serving_llm as _sl + try: + target = resolver() + except _sl.ResolutionError as e: + self._bind_error, self._bind_note = str(e), None + self._bound_at = now + return + except _sl.ResolutionUnavailable as e: + self._bind_to_env(note=str(e)) + self._bound_at = now + return + except Exception as e: # noqa: BLE001 — a resolver bug must not take the health surface down + self._bind_to_env(note=f"unexpected resolution failure: {e}") + self._bound_at = now + return + self._bind_error = None + if target is None: # pointer unset → the configured default base (today's behavior) + self._bind_to_env(note=None) + else: + self.model = os.path.expanduser(target["base_gguf"]) + self.lora = os.path.expanduser(target["adapter_gguf"]) if target["adapter_gguf"] \ + else None + self.alias = target["model_name"] + self.registry_version = target["version"] + self._bind_note = None + self._bound_at = now + + def _bind_to_env(self, note): + self.model, self.lora = self._env_model, self._env_lora + self.alias, self.registry_version = self._env_alias, None + self._bind_error, self._bind_note = None, note + + def bound_identity(self) -> tuple: + """(model_name, registry_version) the NEXT spawn would serve — the reload verb compares + this to loaded_identity() for the idempotent no-op (FR-256).""" + return (self.alias, self.registry_version) + + def loaded_identity(self): + """(model_name, registry_version) of the resident child, or None when never spawned.""" + return (self._loaded["model_name"], self._loaded["registry_version"]) \ + if self._loaded else None # -- lifecycle interface (data-model.md §EngineAdapter) -------------------------------------- def available(self): - """Binary + model prerequisites present AND usable? Surfaces as `unavailable(reason)` (R7) - instead of a spawn crash — the old supervisor found a missing binary only at load time. - Checks executability / real-file (Codex round 7, 018): an existing-but-non-executable binary - or a MODEL that is a directory would otherwise pass here and let the swap target-probe evict - a working holder for a load that then fails.""" + """Binary + resolved artifacts present AND usable? Surfaces as `unavailable(reason)` (R7) + instead of a spawn crash. Rebinds first (TTL-bounded) so the swap/reload target-probe + checks the artifacts that WOULD load — an unresolvable serving-LLM target or a missing + base/adapter GGUF refuses here rather than evicting a working holder (FR-265).""" + self.rebind() + if self._bind_error: + return (False, f"serving-LLM target unresolvable: {self._bind_error}") if not (os.path.isfile(self.llama_bin) and os.access(self.llama_bin, os.X_OK)): return (False, f"llama-server missing or not executable at {self.llama_bin} — " f"build llama.cpp (README)") if not os.path.isfile(self.model): return (False, f"model GGUF not found (or not a file) at {self.model}") + if self.lora and not os.path.isfile(self.lora): + return (False, f"LoRA adapter GGUF not found (or not a file) at {self.lora}") return (True, None) def estimate_vram(self) -> float: - """Quantized weights + KV/context/overhead headroom (the supervisor's `_estimate_vram_gb`). - A missing file → assume near-budget rather than 0, so admission never fails open.""" + """Quantized weights + KV/context/overhead headroom (the supervisor's `_estimate_vram_gb`), + plus the adapter's weights when one is bound. A missing file → assume near-budget rather + than 0, so admission never fails open.""" try: size_gb = os.path.getsize(self.model) / (1024 ** 3) except OSError: return self.vram_budget_gb * 0.95 + if self.lora: + try: + size_gb += os.path.getsize(self.lora) / (1024 ** 3) + except OSError: + return self.vram_budget_gb * 0.95 return size_gb * 1.2 + 1.0 def spawn(self): """Start `llama-server` on a fresh dynamic port; return the Popen (the shared lifecycle - records its pid on admission and drives readiness/teardown).""" + records its pid on admission and drives readiness/teardown). Serves the CURRENT binding — + `-m [--lora ]` (022 US3, FR-263) — and captures it as the loaded identity + the health surface reports (FR-260).""" self._port = free_port() - return subprocess.Popen( - [self.llama_bin, "-m", self.model, "--host", "127.0.0.1", "--port", str(self._port), - "-ngl", self.ngl, "--ctx-size", self.ctx, "--alias", self.alias], - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + cmd = [self.llama_bin, "-m", self.model, "--host", "127.0.0.1", + "--port", str(self._port), "-ngl", self.ngl, "--ctx-size", self.ctx, + "--alias", self.alias] + if self.lora: + cmd += ["--lora", self.lora] + self._loaded = {"model_name": self.alias, "registry_version": self.registry_version, + "base": os.path.basename(self.model), + "adapter": os.path.basename(self.lora) if self.lora else None} + return subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def ready(self) -> bool: if self._port is None: @@ -94,7 +194,8 @@ def ready(self) -> bool: # -- forward surface (engine-specific; the agent calls these while holding the runtime lock) - def forward(self, verb: str, body: dict, load_ms: float) -> dict: """Non-streaming inference. Byte-compatible response with the retired supervisor's `/infer` - ({text, load_ms, infer_ms, model, usage}).""" + ({text, load_ms, infer_ms, model, usage}); `model` names the registry model actually + serving (FR-262 — the loaded identity, not a static env alias).""" if verb != "infer": raise ValueError(f"llm engine has no verb {verb!r}") payload, _ = self._chat_payload(body, stream=False) @@ -107,7 +208,7 @@ def forward(self, verb: str, body: dict, load_ms: float) -> dict: "text": data["choices"][0]["message"]["content"], "load_ms": load_ms, "infer_ms": round((time.perf_counter() - t0) * 1000, 1), - "model": self.alias, + "model": self._served_name(), "usage": data.get("usage", {}), } @@ -120,7 +221,7 @@ def stream(self, verb: str, body: dict, load_ms: float): req = urllib.request.Request(self._url("/v1/chat/completions"), data=payload, headers={"Content-Type": "application/json"}, method="POST") t0 = time.perf_counter() - yield self._frame({"event": "start", "load_ms": load_ms, "model": self.alias}) + yield self._frame({"event": "start", "load_ms": load_ms, "model": self._served_name()}) ntok = 0 with urllib.request.urlopen(req, timeout=300) as r: for raw in r: @@ -139,23 +240,41 @@ def stream(self, verb: str, body: dict, load_ms: float): yield self._frame({"event": "token", "text": delta}) infer_ms = round((time.perf_counter() - t0) * 1000, 1) yield self._frame({"event": "done", "infer_ms": infer_ms, "tokens": ntok, - "model": self.alias}) + "model": self._served_name()}) def health(self, resident: bool) -> dict: - """Byte-compatible with the retired supervisor's `/health` — the shared GPU-tenant payload - (see `_common.engine_health`); gateway `serving.gpu_state` reads - `ok`/`resident`/`model`/`lease_holder`.""" + """The shared GPU-tenant payload (`_common.engine_health`) plus the 022 honest-identity + fields (contracts/agent-identity-and-allowlist.md): `model_name` + `registry_version` name + what is ACTUALLY resident when resident, else what the next load would serve; `base` / + `adapter` carry the resolved artifact identities for a served fine-tune (FR-274).""" ok, reason = self.available() est = self.estimate_vram() if os.path.isfile(self.model) else None - return engine_health(self._admission, ok=ok, reason=reason, resident=resident, - model=self.alias, vram_budget_gb=self.vram_budget_gb, est_vram=est) + ident = self._loaded if (resident and self._loaded) else { + "model_name": self.alias, "registry_version": self.registry_version, + "base": os.path.basename(self.model), + "adapter": os.path.basename(self.lora) if self.lora else None} + payload = engine_health(self._admission, ok=ok, reason=reason, resident=resident, + model=ident["model_name"], vram_budget_gb=self.vram_budget_gb, + est_vram=est) + payload.update({"model_name": ident["model_name"], + "registry_version": ident["registry_version"], + "base": ident["base"], "adapter": ident["adapter"]}) + if self._bind_note: + payload["binding"] = f"registry resolution unavailable — serving the configured " \ + f"default ({self._bind_note})" + return payload # -- helpers --------------------------------------------------------------------------------- + def _served_name(self) -> str: + """The model name for a response: the LOADED identity (a forward always runs against the + resident child), falling back to the binding before any spawn.""" + return self._loaded["model_name"] if self._loaded else self.alias + def _chat_payload(self, body: dict, *, stream: bool): prompt = body.get("prompt", "") max_tokens = int(body.get("max_tokens", 256)) temperature = float(body.get("temperature", 0.7)) - payload = {"model": self.alias, + payload = {"model": self._served_name(), "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens, "temperature": temperature} if stream: diff --git a/hostagent/main.py b/hostagent/main.py index 66c9cf4..fa30a1a 100644 --- a/hostagent/main.py +++ b/hostagent/main.py @@ -308,6 +308,25 @@ def handle_post(path: str, query: str, content_type: str, control_header: str, r result = rt.unload(drain_timeout_s=float(body.get("drain_timeout_s", 10))) code = 200 if result.get("status") in ("unloaded", "idle") else 409 return "json", code, result + if path == "/control/reload": + # 022 T466/T467 (FR-255): the gateway's gated promote requests this so the newly selected + # serving-LLM goes live via a controlled reload — cross-tenant swap or same-tenant + # force-reload, job holders never preempted (the preserved 409 vocabulary). Secret-gated + # like /control/unload (state-changing). + if CONTROL_SECRET and control_header != CONTROL_SECRET: + return "json", 403, {"error": "bad or missing X-Agent-Control"} + body = _parse_json(raw_body) + if body is None: + return "json", 400, {"error": "invalid JSON"} + try: + result = swap_mod.reload_serving_llm( + manager, preempt=bool(body.get("preempt")), + batch_active_fn=lambda: jobs.health_fields()["gpu_batch_active"], + drain_timeout_s=float(body.get("drain_timeout_s", 10))) + except Exception as e: # noqa: BLE001 — mapped to the preserved status vocabulary + code, payload = error_response(e) + return "json", code, payload + return "json", 200, result # Jobs surface (018 T362, contracts/agent-api.md) + the legacy trainer aliases the # gateway still calls (byte-compat, FR-177 — retired from the gateway at US4/cleanup). if path == "/jobs": diff --git a/hostagent/serving_llm.py b/hostagent/serving_llm.py new file mode 100644 index 0000000..e3402f6 --- /dev/null +++ b/hostagent/serving_llm.py @@ -0,0 +1,134 @@ +"""Cold-load resolver for the active serving-LLM (022 T463 — contracts/serving-resolution.md). + +Given the platform's ActiveServingLLM pointer (platformlib.store, written by the gateway's gated +promote), resolve WHAT the llama.cpp engine should serve: + + {model_name, version, kind, base_gguf, adapter_gguf|None, base|None} + + full-model → base_gguf = the version's source, no adapter + lora-adapter → adapter_gguf = the version's source; base_gguf = the registered base's source, + resolved from `base_model` lineage in ONE hop (platformlib.llmresolve) + +Called by the llama adapter on each cold load (`rebind`) — the exact placement the vision/embed/ +tabular children use ("resolve @serving each cold load", research R3), so a promotion is picked up +on the next controlled (re)load with no process restart. + +Failure vocabulary (the seam the adapter + reload verb key on): + - ResolutionError — the pointer names a target that does NOT resolve (no @serving version, + no base, adapter chain, artifact missing). The promote/select is refused + and the currently-served LLM stays unchanged (FR-265). Fail loud. + - ResolutionUnavailable — the pointer store or the registry is UNREACHABLE. The adapter falls + back to the env-configured default base (today's behavior) and surfaces + the degradation in health — the served identity still reports what is + ACTUALLY bound, so honesty (FR-260) is preserved; nothing guesses. + +Artifact locality (research R7): a filesystem source must already exist — nothing is ever +downloaded from outside the box. An `s3://` source (adapters live in the models bucket of the +LOCAL Garage) is materialized once into the state-dir cache; that is same-machine object storage, +so Principle I holds. Stdlib-importable: mlflow/boto3/psycopg load lazily inside resolve() — the +agent's default transport carries no pip deps; the training venv it runs under has all three. +""" +import hashlib +import os + +from platformlib import llmresolve +from platformlib.topology import STATE_DIR + + +class ResolutionError(Exception): + """The active target is invalid — refuse the promote/select, served LLM unchanged (FR-265).""" + + +class ResolutionUnavailable(Exception): + """Pointer store / registry unreachable — the caller may fall back to the configured default + (surfaced, never silent).""" + + +#: Where s3:// adapter artifacts are materialized (keyed by source digest — a re-promote of the +#: same version reuses the cached file; a new version is a new source → a new cache entry). +CACHE_DIR = os.path.join(STATE_DIR, "llm-artifacts") + + +def active_model_name(store=None): + """The pointer's model name, or None when unset (⇒ the configured default base serves). + `store` is injectable for tests (defaults to platformlib.store).""" + if store is None: + from platformlib import store as _default_store + store = _default_store + try: + conn = store.connect() + try: + rec = store.get_serving_llm(conn) + finally: + conn.close() + except store.StoreError as e: + raise ResolutionUnavailable(f"serving-LLM pointer unreadable: {e}") from e + return rec["model_name"] if rec else None + + +def resolve(*, store=None, client=None, materialize=None): + """The full cold-load resolution. Returns None when the pointer is unset; raises + ResolutionError / ResolutionUnavailable per the module contract.""" + name = active_model_name(store) + if name is None: + return None + if client is None: + try: + from mlflow.tracking import MlflowClient + client = MlflowClient( + tracking_uri=os.getenv("MLFLOW_TRACKING_URI", "http://localhost:5500")) + except Exception as e: # noqa: BLE001 — no mlflow in this interpreter + raise ResolutionUnavailable(f"mlflow client unavailable: {e}") from e + try: + rec = llmresolve.resolve_serving_record(client, name) + except llmresolve.LLMResolutionError as e: + raise ResolutionError(str(e)) from e + except Exception as e: # noqa: BLE001 — registry down/unreachable, not an invalid target + raise ResolutionUnavailable(f"registry unreachable resolving {name!r}: {e}") from e + mat = materialize or _materialize + if rec["kind"] == llmresolve.KIND_ADAPTER: + base_gguf = mat(rec["base"]["source"]) + adapter_gguf = mat(rec["source"]) + else: + base_gguf = mat(rec["source"]) + adapter_gguf = None + return {"model_name": rec["model_name"], "version": rec["version"], "kind": rec["kind"], + "base_gguf": base_gguf, "adapter_gguf": adapter_gguf, "base": rec["base"]} + + +def _materialize(source: str) -> str: + """A local file path for `source`. Filesystem sources must already exist (the operator-curated + zoo — FR-265, nothing auto-downloaded); s3:// sources fetch once from the local Garage.""" + if str(source).startswith("s3://"): + return _fetch_s3(str(source)) + path = os.path.expanduser(str(source)) + if not os.path.isfile(path): + raise ResolutionError( + f"artifact not found (or not a file) at {path} — the base/adapter GGUF must already " + f"be present locally (nothing is auto-downloaded, FR-265)") + return path + + +def _fetch_s3(source: str) -> str: + bucket, _, key = source[len("s3://"):].partition("/") + if not bucket or not key: + raise ResolutionError(f"malformed artifact source {source!r}") + cache = os.path.join(os.path.expanduser(CACHE_DIR), + hashlib.sha256(source.encode()).hexdigest()[:16] + ".gguf") + if os.path.isfile(cache) and os.path.getsize(cache) > 0: + return cache + try: + from platformlib.store import s3_client + os.makedirs(os.path.dirname(cache), exist_ok=True) + tmp = cache + ".part" + s3_client().download_file(bucket, key, tmp) + os.replace(tmp, cache) + except ResolutionError: + raise + except Exception as e: # noqa: BLE001 — classify: object gone = invalid target; else outage + code = str(getattr(e, "response", {}).get("Error", {}).get("Code", "")) \ + if hasattr(e, "response") else "" + if code in ("404", "NoSuchKey") or "Not Found" in str(e) or "NoSuchKey" in str(e): + raise ResolutionError(f"artifact object missing in the store: {source}") from e + raise ResolutionUnavailable(f"object store unreachable fetching {source}: {e}") from e + return cache diff --git a/hostagent/swap.py b/hostagent/swap.py index 3d85fc7..fdb985f 100644 --- a/hostagent/swap.py +++ b/hostagent/swap.py @@ -100,3 +100,73 @@ def preempt_for(manager, target_engine_id: str, drain_timeout_s: float = 10.0, * return {"swapped": True, "evicted": holder["tenant"], "load_ms": load_ms} finally: admission.end_swap(target_engine_id) # no-op after a rollback retarget + + +def reload_serving_llm(manager, *, preempt: bool = False, batch_active_fn=None, + drain_timeout_s: float = 10.0) -> dict: + """Make the ACTIVE serving-LLM live now (022 T466, FR-255 — contracts/serving-resolution.md). + + Admission tenancy is per-ENGINE, so a served-LLM switch has two cases: + + - **cross-tenant** (a non-LLM engine holds the GPU): `preempt_for` — evict → free → load, + with the job-holder refusal and the target-probe it already carries; + - **same-tenant model switch** (the `llm` engine is resident with a DIFFERENT model — the + common US1 case, where `preempt_for` is a satisfied no-op): an explicit force-reload, + unload the llm child → `ensure_loaded()`, so `llama-server` re-spawns with the newly + resolved base+adapter. Reusing preempt_for alone here silently keeps serving the old + GGUF (spec review PR #64 §1). + + Sequencing guards, in order: fresh resolve (rebind(force=True) busts the adapter's TTL) → + target-probe (`available()`, so a bad artifact never evicts a working holder, FR-256/257) → + idempotent no-op when the resolved model+version is already resident → job/batch holders are + NEVER displaced (FR-259) → any displacement of a resident serving model requires the + operator's confirm (`preempt=true`, set by the console's ConfirmDialog — FR-258). Both paths + are strictly sequential evict→load under admission: never two models resident (SC-147). + """ + rt = manager.runtimes.get("llm") + if rt is None: + raise PreemptRefused("no llm engine registered") + adapter = rt.adapter + adapter.rebind(force=True) + ok, reason = adapter.available() + if not ok: # probe BEFORE any unload/evict — the working holder stays put (FR-265) + raise PreemptRefused(f"serving-LLM target not loadable: {reason}") + holder = manager.admission.holder() + if holder is None: + return {"status": "loaded", "load_ms": rt.ensure_loaded(), + **_identity(adapter)} + if holder["tenant"] != "llm": + if holder["kind"] in NON_PREEMPTABLE_KINDS: + raise PreemptRefused( + f"{holder['tenant']} is running a job (training/HPO/batch) — never preempted; " + f"the switch is deferred to the next load (FR-259)") + if not preempt: + raise PreemptRefused( + f"would displace resident {holder['tenant']} — operator confirmation required " + f"(re-issue with preempt=true)") + res = preempt_for(manager, "llm", drain_timeout_s, batch_active_fn=batch_active_fn) + return {"status": "swapped", "evicted": res["evicted"], "load_ms": res["load_ms"], + **_identity(adapter)} + # same-tenant: the llm engine itself holds the slot + if holder["kind"] in NON_PREEMPTABLE_KINDS: + raise PreemptRefused("the llm slot is held by a job — never preempted (FR-259)") + if batch_active_fn is not None and batch_active_fn(): + raise PreemptRefused("a GPU batch is driving the llm engine — not reloaded (FR-155)") + if adapter.loaded_identity() == adapter.bound_identity(): + return {"status": "noop", "load_ms": 0.0, **_identity(adapter)} # idempotent (FR-256) + if not preempt: + loaded = adapter.loaded_identity() + raise PreemptRefused( + f"would displace the resident LLM {loaded[0] if loaded else '?'} — operator " + f"confirmation required (re-issue with preempt=true)") + result = rt.unload(drain_timeout_s=drain_timeout_s) + if result.get("status") not in ("unloaded", "idle"): + raise SwapError(f"llm did not unload for the model switch: " + f"{result.get('detail') or result}") + return {"status": "reloaded", "load_ms": rt.ensure_loaded(), **_identity(adapter)} + + +def _identity(adapter) -> dict: + """The bound (now loading/loaded) serving identity for the reload response.""" + name, version = adapter.bound_identity() + return {"model_name": name, "registry_version": version} diff --git a/infra/postgres/init.sql b/infra/postgres/init.sql index ff3a188..e0112b2 100644 --- a/infra/postgres/init.sql +++ b/infra/postgres/init.sql @@ -69,3 +69,11 @@ CREATE TABLE IF NOT EXISTS suggestions ( resolved_at timestamptz, actor text ); + +-- 022 T461: the single active serving-LLM pointer (one row; absent => the default base serves). +CREATE TABLE IF NOT EXISTS serving_llm ( + singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton), + model_name text NOT NULL, + selected_at timestamptz NOT NULL, + selected_by text NOT NULL +); diff --git a/platformlib/llmresolve.py b/platformlib/llmresolve.py new file mode 100644 index 0000000..4aa60f5 --- /dev/null +++ b/platformlib/llmresolve.py @@ -0,0 +1,141 @@ +"""Shared serving-LLM registry walk (022 T463/T464 — contracts/serving-resolution.md). + +ONE implementation of "active model name → its `@serving` version → the base/adapter record", +used by BOTH resolution sites so they cannot drift: + + - the host agent's cold-load resolver (`hostagent/serving_llm.py`) — adds the pointer read and + artifact materialization on top of this walk; + - the gateway's promote pre-check + task surfaces (`gateway/app/registry.py`) — refuses an + unresolvable promotion BEFORE the alias moves (FR-265) and exposes base-vs-adapter kind. + +They can't share an import path any other way: the gateway image ships only `gateway/app` + +`platformlib` (gateway/Dockerfile), and the agent's default transport is stdlib-only — so this +module is dependency-free and takes a duck-typed MLflow client (`get_model_version_by_alias`, +`search_model_versions`) rather than importing mlflow. Client/connection exceptions PROPAGATE +(the callers classify unreachable-vs-invalid); only "does not exist" is normalized here. + +Kind rules (data-model.md §LLMVersion): + - `kind=full-model` → served standalone (`-m source`). + - `kind=lora-adapter` → served on its base (`-m base --lora source`); `base_model` MUST resolve + DIRECTLY to a registered full-model version — an adapter pointing at another adapter is a + resolution error, not a multi-hop walk (spec review PR #64, R2 note). + - Legacy versions without `kind` (pre-022, FR-267) are inferred: a `base_model` + `format=gguf` + pair is an adapter (the fine-tune flow's historical shape); anything else standalone. +""" + +SERVING_ALIAS = "serving" +TASK_TAG = "task" +ENGINE_TAG = "serving_engine" +KIND_TAG = "kind" +BASE_MODEL_TAG = "base_model" +BASE_ID_TAG = "base_id" # a registered BASE's raw HF/local id (scripts/register_base_gguf) +FORMAT_TAG = "format" + +KIND_FULL = "full-model" +KIND_ADAPTER = "lora-adapter" +TEXT_GENERATION = "text-generation" +LLAMA_ENGINE = "llama.cpp" + + +class LLMResolutionError(Exception): + """The target does not resolve (no @serving / no base / adapter chain) — refuse the + promote/select with this reason and leave the served LLM unchanged (FR-265).""" + + +def _escape(value: str) -> str: + """Single-quote-escape a value for an MLflow filter string (the registry.py house rule).""" + return str(value).replace("'", "''") + + +def _is_not_found(exc: Exception) -> bool: + """True when an MLflow error means 'does not exist' (vs. unreachable). Duck-typed on the REST + error code so this module needs no mlflow import.""" + if getattr(exc, "error_code", None) == "RESOURCE_DOES_NOT_EXIST": + return True + text = str(exc) + return "RESOURCE_DOES_NOT_EXIST" in text or "not found" in text.lower() + + +def _by_alias(client, name: str): + """The @serving model version of `name`, or None when the model/alias doesn't exist. + Connection errors propagate for the caller to classify.""" + try: + return client.get_model_version_by_alias(name, SERVING_ALIAS) + except Exception as e: # noqa: BLE001 — only not-found is normalized; the rest propagates + if _is_not_found(e): + return None + raise + + +def effective_kind(tags: dict) -> str: + """A version's base-vs-adapter kind, inferring legacy shapes (FR-267): an explicit + `kind` wins; else `base_model`+`format=gguf` (the pre-022 fine-tune stamp) is an adapter; + anything else serves standalone.""" + kind = (tags or {}).get(KIND_TAG) + if kind in (KIND_FULL, KIND_ADAPTER): + return kind + if (tags or {}).get(BASE_MODEL_TAG) and (tags or {}).get(FORMAT_TAG) == "gguf": + return KIND_ADAPTER + return KIND_FULL + + +def task_from_kind(tags: dict): + """Defense-in-depth task inference (022 T478, FR-267): a LoRA adapter or any GGUF artifact is + a text-generation version even when the `task` tag is missing (legacy, pre-backfill) — GGUF is + llama.cpp's format, so the inference is unambiguous. Returns the task or None (no guess for + non-LLM shapes).""" + tags = tags or {} + if tags.get(KIND_TAG) == KIND_ADAPTER or tags.get(FORMAT_TAG) == "gguf": + return TEXT_GENERATION + return None + + +def resolve_base_version(client, base_ref: str): + """Map an adapter's `base_model` reference to a registered full-model version — SINGLE hop + (contracts/serving-resolution.md). `base_ref` is either a registered model name or the raw + HF/local id the trainer stamps (matched via the base's `base_id` tag). Raises + LLMResolutionError when nothing full-model matches (including the adapter-chain case).""" + candidates = [] + aliased = _by_alias(client, base_ref) + if aliased is not None: + candidates.append(aliased) + named = client.search_model_versions(f"name='{_escape(base_ref)}'") + candidates += sorted(named, key=lambda m: int(m.version), reverse=True) + if not candidates: # not a registered name → the trainer's raw id, matched by base_id tag + tagged = client.search_model_versions(f"tags.{BASE_ID_TAG}='{_escape(base_ref)}'") + candidates = sorted(tagged, key=lambda m: int(m.version), reverse=True) + for mv in candidates: + if effective_kind(dict(mv.tags or {})) == KIND_FULL: + return mv + if candidates: + raise LLMResolutionError( + f"base {base_ref!r} resolves only to adapter versions — an adapter's base must be a " + f"registered full-model version (no adapter chains)") + raise LLMResolutionError( + f"base {base_ref!r} is not a registered full-model text-generation version — register the " + f"local base GGUF first (scripts/register_base_gguf.py)") + + +def resolve_serving_record(client, model_name: str) -> dict: + """`model_name` → its @serving version → the servable record: + {model_name, version, kind, source, tags, base:{name, version, source}|None}. + + Pure registry read — no file checks (the agent layers artifact materialization on top). + Raises LLMResolutionError when the @serving alias or an adapter's base doesn't resolve.""" + mv = _by_alias(client, model_name) + if mv is None: + raise LLMResolutionError( + f"{model_name!r} has no @{SERVING_ALIAS} version — nothing is promoted to serve") + tags = dict(mv.tags or {}) + kind = effective_kind(tags) + base = None + if kind == KIND_ADAPTER: + base_ref = tags.get(BASE_MODEL_TAG) + if not base_ref: + raise LLMResolutionError( + f"{model_name} v{mv.version} is a LoRA adapter with no base_model lineage — " + f"its base cannot be resolved (FR-264/265)") + base_mv = resolve_base_version(client, base_ref) + base = {"name": base_mv.name, "version": str(base_mv.version), "source": base_mv.source} + return {"model_name": mv.name, "version": str(mv.version), "kind": kind, + "source": mv.source, "tags": tags, "base": base} diff --git a/platformlib/store.py b/platformlib/store.py index dee76d7..4d70352 100644 --- a/platformlib/store.py +++ b/platformlib/store.py @@ -171,10 +171,21 @@ def list_common_prefixes(s3, bucket: str, prefix: str = "", delimiter: str = "/" resolved_at timestamptz, actor text ); + +-- 022 T461 (data-model.md §ActiveServingLLM): the ONE platform-scoped pointer naming the active +-- text-generation model. A single row (the CHECKed boolean PK makes a second row impossible); +-- absent row ⇒ the configured default base serves (today's behavior). Additive — no version bump. +CREATE TABLE IF NOT EXISTS serving_llm ( + singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton), + model_name text NOT NULL, + selected_at timestamptz NOT NULL, + selected_by text NOT NULL +); """ #: The tables bootstrap() must create — used by the test + a post-bootstrap self-check. -TABLES = ("meta", "predictions", "labels", "capture_index", "jobs", "policies", "suggestions") +TABLES = ("meta", "predictions", "labels", "capture_index", "jobs", "policies", "suggestions", + "serving_llm") class StoreError(Exception): @@ -526,6 +537,45 @@ def resolve_suggestion(conn, suggestion_id: str, state: str, actor: str, resolve return _suggestion_row(row) if row else None +# ================================================================================================== +# 022 T461 — the ActiveServingLLM pointer (data-model.md §ActiveServingLLM) +# ================================================================================================== +# +# One platform-scoped row naming the active text-generation model. Written by the gateway's gated +# promote (022 US1 — promote = go live); read by the host agent's cold-load resolver +# (hostagent/serving_llm.py). Unset ⇒ the configured default base serves (today's behavior), so a +# fresh store changes nothing until an operator promotes an LLM. Timestamps follow the house rule: +# epoch floats at the seam, timestamptz in the column. + + +def set_serving_llm(conn, model_name: str, selected_at, selected_by: str) -> None: + """Point the active serving-LLM at `model_name` (upsert — the singleton PK keeps it one row).""" + with conn.cursor() as cur: + cur.execute( + "INSERT INTO serving_llm (singleton, model_name, selected_at, selected_by) " + "VALUES (true, %s, %s, %s) ON CONFLICT (singleton) DO UPDATE SET " + "model_name = EXCLUDED.model_name, selected_at = EXCLUDED.selected_at, " + "selected_by = EXCLUDED.selected_by", + (model_name, _dt(selected_at), selected_by)) + + +def get_serving_llm(conn): + """The active serving-LLM selection {model_name, selected_at, selected_by}, or None when unset + (⇒ the configured default base serves).""" + with conn.cursor() as cur: + cur.execute("SELECT model_name, selected_at, selected_by FROM serving_llm") + row = cur.fetchone() + if row is None: + return None + return {"model_name": row[0], "selected_at": _epoch(row[1]), "selected_by": row[2]} + + +def clear_serving_llm(conn) -> None: + """Unset the pointer — back to the configured default base (tests/reset; no operator surface).""" + with conn.cursor() as cur: + cur.execute("DELETE FROM serving_llm") + + # ================================================================================================== # US2 job records (US4 T375-B) — the durable `jobs` table behind the host-agent journal # ================================================================================================== diff --git a/scripts/backfill_llm_task_tags.py b/scripts/backfill_llm_task_tags.py new file mode 100644 index 0000000..649c84a --- /dev/null +++ b/scripts/backfill_llm_task_tags.py @@ -0,0 +1,74 @@ +"""Backfill task descriptors onto legacy text-generation versions (022 T477 — FR-267). + +Fine-tunes registered before 022 carry `kind=lora-adapter`/`format=gguf` but NO `task` tag, so the +console showed them as "no renderer" placeholders and the serving surfaces couldn't route them +(the ops-bot-v1/v2 gap observed live). This one-time backfill stamps what T476 now writes at +registration: + + task=text-generation (identified by kind=lora-adapter OR format=gguf — GGUF is + serving_engine=llama.cpp llama.cpp's format, so the inference is unambiguous) + +Idempotent + NON-CLOBBER: an existing `task` or `serving_engine` tag is never overwritten — a +re-run changes nothing (the report's `skipped` count proves it). Read + tag only; no alias moves, +no artifact writes. + +Usage: python scripts/backfill_llm_task_tags.py (host side; MLFLOW_TRACKING_URI to override) +""" +import os +import sys + +MLFLOW_URI = os.getenv("MLFLOW_TRACKING_URI", "http://localhost:5500") + + +def backfill(client, log=print) -> dict: + """Stamp missing task/serving_engine tags on every legacy LLM-shaped version. Returns + {"tagged": [...], "skipped": [...]} — separated from main() so idempotency and the non-clobber + rule are unit-testable against a fake client.""" + from platformlib import llmresolve + + # Two searches because MLflow filter strings have no OR: adapters by kind, bases/full GGUFs by + # format. De-duped by (name, version). + seen, candidates = set(), [] + for query in (f"tags.{llmresolve.KIND_TAG}='{llmresolve.KIND_ADAPTER}'", + f"tags.{llmresolve.FORMAT_TAG}='gguf'"): + for mv in client.search_model_versions(query): + key = (mv.name, str(mv.version)) + if key not in seen: + seen.add(key) + candidates.append(mv) + + report = {"tagged": [], "skipped": []} + for mv in candidates: + tags = dict(mv.tags or {}) + if llmresolve.task_from_kind(tags) != llmresolve.TEXT_GENERATION: + report["skipped"].append({"name": mv.name, "version": str(mv.version), + "reason": "not an LLM artifact shape"}) + continue + missing = {} + if not tags.get(llmresolve.TASK_TAG): # non-clobber: only ever fill a missing tag + missing[llmresolve.TASK_TAG] = llmresolve.TEXT_GENERATION + if not tags.get(llmresolve.ENGINE_TAG): + missing[llmresolve.ENGINE_TAG] = llmresolve.LLAMA_ENGINE + if not missing: + report["skipped"].append({"name": mv.name, "version": str(mv.version), + "reason": "already tagged"}) + log(f"= {mv.name} v{mv.version}: already tagged") + continue + for key, value in missing.items(): + client.set_model_version_tag(mv.name, str(mv.version), key, value) + report["tagged"].append({"name": mv.name, "version": str(mv.version), + "added": sorted(missing)}) + log(f"+ {mv.name} v{mv.version}: {', '.join(f'{k}={v}' for k, v in missing.items())}") + return report + + +def main() -> int: + from mlflow.tracking import MlflowClient + + report = backfill(MlflowClient(tracking_uri=MLFLOW_URI)) + print(f"done: {len(report['tagged'])} tagged, {len(report['skipped'])} skipped") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/register_base_gguf.py b/scripts/register_base_gguf.py new file mode 100644 index 0000000..bf05f2f --- /dev/null +++ b/scripts/register_base_gguf.py @@ -0,0 +1,83 @@ +"""Register the local base GGUFs as first-class registry records (022 T462 — research R2/R7). + +The serving resolver (hostagent/serving_llm.py) maps a fine-tune's `base_model` lineage to a +**registered** `kind=full-model` text-generation version and serves `-m --lora `. +That only works if the bases in the local zoo (`~/models/gguf/`) exist as registry records — this +script creates them: one `kind=full-model` version per base, `source` pointing at the local GGUF, +tagged so both the name-match and the `base_id`-match resolution paths hit: + + - `task=text-generation` + `serving_engine=llama.cpp` (009 routing metadata), + - `kind=full-model` + `format=gguf` (the 022 base-vs-adapter discriminator), + - `base_id=` — the raw base string the trainer stamps into an adapter's `base_model` + (e.g. `Qwen/Qwen2.5-0.5B-Instruct`), so lineage written before 022 resolves too. + +Idempotent: a base whose (name, source) is already registered is skipped — re-runs register +nothing new. A base GGUF absent from the zoo is skipped with a note (a 16 GB-budget box may carry +only one base; Principle I/III — nothing is downloaded here, ever). + +Usage: python scripts/register_base_gguf.py (host side; MLFLOW_TRACKING_URI/GGUF_DIR to override) +""" +import os +import sys + +MLFLOW_URI = os.getenv("MLFLOW_TRACKING_URI", "http://localhost:5500") +GGUF_DIR = os.path.expanduser(os.getenv("GGUF_DIR", "~/models/gguf")) + +# The local zoo (research R7 — bounded, operator-curated). `name` is the registered-model name the +# platform serves under (the 7B matches SERVING_MODEL); `base_id` is the HF id adapters record as +# `base_model`; `file` follows the zoo naming (the 0.5B f16 comes from scoring's resolve_base_gguf). +BASES = [ + {"name": "qwen2.5-7b-instruct-q4_k_m", + "base_id": "Qwen/Qwen2.5-7B-Instruct", + "file": "Qwen2.5-7B-Instruct-Q4_K_M.gguf"}, + {"name": "qwen2.5-0.5b-instruct", + "base_id": "Qwen/Qwen2.5-0.5B-Instruct", + "file": "Qwen_Qwen2.5-0.5B-Instruct-f16.gguf"}, +] + + +def register_bases(client, bases=None, gguf_dir=None, log=print) -> dict: + """Register each present base GGUF once. Returns {"registered": [...], "skipped": [...]} — + separated from main() so the idempotency is unit-testable against a fake client.""" + from mlflow.exceptions import MlflowException + + gguf_dir = gguf_dir or GGUF_DIR + report = {"registered": [], "skipped": []} + for base in bases or BASES: + path = os.path.join(gguf_dir, base["file"]) + if not os.path.isfile(path): + report["skipped"].append({"name": base["name"], "reason": f"no GGUF at {path}"}) + log(f"~ {base['name']}: no GGUF at {path} — skipped (nothing is downloaded)") + continue + try: + existing = client.search_model_versions(f"name='{base['name']}'") + except MlflowException: + existing = [] + if any(mv.source == path for mv in existing): + report["skipped"].append({"name": base["name"], "reason": "already registered"}) + log(f"= {base['name']}: already registered from {path}") + continue + try: + client.create_registered_model(base["name"]) + except MlflowException: + pass # exists — adding a version is fine + mv = client.create_model_version( + name=base["name"], source=path, run_id=None, + tags={"kind": "full-model", "format": "gguf", "task": "text-generation", + "serving_engine": "llama.cpp", "base_id": base["base_id"]}) + report["registered"].append({"name": base["name"], "version": str(mv.version), + "source": path}) + log(f"+ {base['name']} v{mv.version} <- {path}") + return report + + +def main() -> int: + from mlflow.tracking import MlflowClient + + report = register_bases(MlflowClient(tracking_uri=MLFLOW_URI)) + print(f"done: {len(report['registered'])} registered, {len(report['skipped'])} skipped") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/specs/022-registry-driven-llm-serving/tasks.md b/specs/022-registry-driven-llm-serving/tasks.md index 992183e..bea3bcc 100644 --- a/specs/022-registry-driven-llm-serving/tasks.md +++ b/specs/022-registry-driven-llm-serving/tasks.md @@ -20,12 +20,12 @@ test may regress — especially the Principle II admission tests. ## Phase 1: Setup (shared, mechanical — unblocks everything) -- [ ] **T461** Add the **ActiveServingLLM** pointer + served-identity persistence to the relational +- [X] **T461** Add the **ActiveServingLLM** pointer + served-identity persistence to the relational store: schema + accessors in `platformlib/store` (get/set the active text-generation `model_name`, with `selected_at`/`selected_by`; unset ⇒ resolve to the configured default base). Per [data-model.md](./data-model.md). Validate: get/set round-trips; a fresh store resolves to the default base; `pytest` covers the accessor. -- [ ] **T462** [P] `scripts/register_base_gguf.py` — register the local base GGUFs (Qwen 0.5B / 7B in +- [X] **T462** [P] `scripts/register_base_gguf.py` — register the local base GGUFs (Qwen 0.5B / 7B in `~/models/gguf/`) as `kind=full-model` text-generation registry versions (`task=text-generation`, `serving_engine=llama.cpp`, `source` → the local GGUF), so a fine-tune's `base_model` lineage resolves to a first-class registry record (research R2/R7). Idempotent (re-run registers nothing @@ -34,14 +34,14 @@ test may regress — especially the Principle II admission tests. ## Phase 2: Foundational (blocking prerequisite for resolution + the live switch) -- [ ] **T463** `hostagent/serving_llm.py` — the cold-load resolver (research R1/R2/R3): given the +- [X] **T463** `hostagent/serving_llm.py` — the cold-load resolver (research R1/R2/R3): given the ActiveServingLLM pointer, resolve its `@serving` version → `{model_name, version, base_gguf, adapter_gguf|null}` per [contracts/serving-resolution.md](./contracts/serving-resolution.md): `full-model` → `base_gguf=source`; `lora-adapter` → `adapter_gguf=source` + `base_gguf=resolve(base_model → registered base version.source)`. Raise a clear resolution error on missing `@serving`/artifact/ base. Pure read of {store pointer, MLflow}; no network fetch. Unit-test against a fake registry/store (full-model, adapter, missing-base, unset-pointer→default). Blocks US1/US3. -- [ ] **T464** Gateway `registry.py` — active-serving-LLM pointer get/set (store-backed, via T461); +- [X] **T464** Gateway `registry.py` — active-serving-LLM pointer get/set (store-backed, via T461); `resolve_serving_target` stamps/reads `task` for LLM versions and resolves an adapter's base; expose a version's base-vs-adapter `kind` + lineage in the models/tasks surfaces. Unit-test the resolution + pointer round-trip. Blocks US1/US4. @@ -54,7 +54,7 @@ test may regress — especially the Principle II admission tests. controlled reload — no `.env` edit, no manual restart (FR-254/255). **Independent test**: [quickstart.md](./quickstart.md) §US1. -- [ ] **T465** [US1] `hostagent/adapters/llama.py` — call the T463 resolver in `spawn()` / +- [X] **T465** [US1] `hostagent/adapters/llama.py` — call the T463 resolver in `spawn()` / `ensure_loaded()` to bind `self.model` (base) + `self.lora` (adapter, or None) from the registry before launching `llama-server` (spawn already appends `--lora` when set — commit `c28ca97`). This **supersedes** the static `MODEL`/`LORA` env knob (env becomes a fallback/default only). Also set @@ -63,7 +63,7 @@ controlled reload — no `.env` edit, no manual restart (FR-254/255). — so the `/infer` response's `model` matches the registry record (FR-262; spec review PR #64 §2). `available()` verifies the resolved artifacts. Validate: with a promoted version, a cold load serves it and the response `model` names it; `pytest` covers resolve→spawn arg construction + the alias. -- [ ] **T466** [US1] Reload-on-select under admission — TWO cases (admission tenancy is per-engine, +- [X] **T466** [US1] Reload-on-select under admission — TWO cases (admission tenancy is per-engine, not per-model; spec review PR #64 §1): **cross-tenant** (non-LLM/job holder) reuses `hostagent/swap.py:preempt_for` (evict → free → load); **same-tenant model switch** (the `llm` engine already resident with a *different* model — where `preempt_for` is a no-op) performs an @@ -71,7 +71,7 @@ controlled reload — no `.env` edit, no manual restart (FR-254/255). resolved artifact (FR-255 immediate / SC-144). Idempotent no-op only when the resolved model+version is already resident; target-probed before any unload/evict (FR-256/257). `pytest` covers the same-tenant force-reload path (promote B while A resident → B actually loads). -- [ ] **T467** [US1] Make the **gated promote itself the go-live action** (Clarifications 2026-07-05: +- [X] **T467** [US1] Make the **gated promote itself the go-live action** (Clarifications 2026-07-05: promote = go live, one action — no separate "select" gesture): promoting a text-generation version through `POST models/:name/promote` (011/015) MUST also write the ActiveServingLLM pointer (T464) and request the T466 immediate agent reload. Prefer extending the existing promote path over a new @@ -79,6 +79,7 @@ controlled reload — no `.env` edit, no manual restart (FR-254/255). ([contracts/agent-identity-and-allowlist.md](./contracts/agent-identity-and-allowlist.md)). - [ ] **T468** [US1] Validate US1 end-to-end against quickstart §US1 (promote A → `/infer` is A; promote B → `/infer` is B; no host-level action); backend `pytest` (resolver + reload wiring) green. + *(offline half done: resolver + reload wiring `pytest` green; the live quickstart §US1 browser drill runs with the stack up.)* **Checkpoint**: US1 alone makes the LLM console-operable — the core value. @@ -90,18 +91,19 @@ controlled reload — no `.env` edit, no manual restart (FR-254/255). actually resident, so monitoring scores the right model (FR-260/261/262). **Independent test**: quickstart §US2. -- [ ] **T469** [US2] Agent reports the actually-loaded `model_name` + `registry_version` (from T463 +- [X] **T469** [US2] Agent reports the actually-loaded `model_name` + `registry_version` (from T463 resolution) in `/engines/llm/health` and the aggregate health ([contracts/agent-identity-and-allowlist.md](./contracts/agent-identity-and-allowlist.md)). -- [ ] **T470** [US2] Gateway `serving.py gpu_state` (`GET /serving/state`) consumes the agent-reported +- [X] **T470** [US2] Gateway `serving.py gpu_state` (`GET /serving/state`) consumes the agent-reported `model_name` + `registry_version` instead of the fixed `SERVING_MODEL`; degrades to `unknown` when the agent is unreachable (never a stale config guess). -- [ ] **T471** [US2] Prediction logging in `gateway/app/routers/infer.py` + `stream.py` attributes each +- [X] **T471** [US2] Prediction logging in `gateway/app/routers/infer.py` + `stream.py` attributes each served prediction to the agent-reported identity (not `SERVING_MODEL`), so the 013 quality window keys on the correct model+version. No prediction logged under a model the agent is not serving. - [ ] **T472** [US2] Validate US2 against quickstart §US2 (serving-state + `/infer` `model` + logged prediction all agree; a quality check scores the right model+version — the live divergence bug is gone); `pytest` identity tests green. + *(offline half done: identity `pytest` green; the live quickstart §US2 drill runs with the stack up.)* --- @@ -111,14 +113,15 @@ actually resident, so monitoring scores the right model (FR-260/261/262). back to a base restores base behavior (FR-263/264/265). **Independent test**: quickstart §US3. -- [ ] **T473** [US3] Adapter path in the T463 resolver: `lora-adapter` version → `base_model` lineage +- [X] **T473** [US3] Adapter path in the T463 resolver: `lora-adapter` version → `base_model` lineage → registered base version `source` → spawn `-m --lora `. Unit-test the base resolution + arg construction (adapter, chained-parent, base-is-another-registered-version). -- [ ] **T474** [US3] Refuse-on-unresolvable-base (FR-265): a resolution error (missing/absent base or +- [X] **T474** [US3] Refuse-on-unresolvable-base (FR-265): a resolution error (missing/absent base or artifact) refuses the promote/select with a clear reason and leaves the currently-served LLM unchanged — never a wedged/empty serving state. `pytest` covers the refusal. - [ ] **T475** [US3] Validate US3 against quickstart §US3 (promote fine-tune → trained behavior serves; promote base → gone; unresolvable base → refused, unchanged); `pytest` green. + *(offline half done: adapter-path + refusal `pytest` green; the live quickstart §US3 drill runs with the stack up.)* --- @@ -128,17 +131,17 @@ back to a base restores base behavior (FR-263/264/265). selectable LLM panels; legacy untagged versions are backfilled (FR-266/267/268/270). **Independent test**: quickstart §US4. -- [ ] **T476** [US4] `training/flows/finetune.py` — stamp `task=text-generation`, +- [X] **T476** [US4] `training/flows/finetune.py` — stamp `task=text-generation`, `serving_engine=llama.cpp`, and base/parent lineage at registration (mirrors vision/embed tagging), so a new fine-tune registers with a non-null task. `pytest` covers the registration tags. -- [ ] **T477** [P] [US4] `scripts/backfill_llm_task_tags.py` — backfill legacy untagged +- [X] **T477** [P] [US4] `scripts/backfill_llm_task_tags.py` — backfill legacy untagged text-generation versions (identified by `kind=lora-adapter`/`format=gguf`) with `task`/`serving_engine`; idempotent + non-clobber (never overwrites an existing tag). Validate on `ops-bot-v1/v2`: they become `task=text-generation`; re-run is a no-op. -- [ ] **T478** [P] [US4] Gateway resolve fallback: infer `task` from `kind` (lora-adapter/gguf → +- [X] **T478** [P] [US4] Gateway resolve fallback: infer `task` from `kind` (lora-adapter/gguf → text-generation) as defense-in-depth in `resolve_serving_target`/`serving/tasks`, so no valid LLM version is stranded as `task:null` even before the backfill runs. -- [ ] **T479** [US4] Surface a version's base-vs-adapter `kind` + lineage in `GET /serving/tasks` and +- [X] **T479** [US4] Surface a version's base-vs-adapter `kind` + lineage in `GET /serving/tasks` and the models detail so the console renders a working LLM panel (not "no renderer") and shows what would be promoted (FR-268/270). **Also make `serving/tasks` authoritative to the active-serving-LLM pointer (FR-276; spec review PR #64 §3):** `registry.list_tasks()` emits one row per model with ANY @@ -149,6 +152,7 @@ selectable LLM panels; legacy untagged versions are backfilled (FR-266/267/268/2 live LLM, no stale duplicate. `pytest` covers the two-promoted-LLMs dedup. - [ ] **T480** [US4] Validate US4 against quickstart §US4 (new fine-tune task-tagged; backfill idempotent + selectable; promoted fine-tune renders a working panel); `pytest` green. + *(offline half done: tagging/backfill/dedup `pytest` green; the live quickstart §US4 drill runs with the stack up.)* --- @@ -158,16 +162,16 @@ selectable LLM panels; legacy untagged versions are backfilled (FR-266/267/268/2 preempt of a serving holder, job never preempted (FR-257/258/259). **Independent test**: quickstart §US5. -- [ ] **T481** [US5] Switch under admission: displacing a resident **serving** holder goes through the +- [X] **T481** [US5] Switch under admission: displacing a resident **serving** holder goes through the operator-confirmed swap; a **job** holder (training/HPO/batch) is refused/deferred with the existing 409 reason (never preempted); never two-resident. Reuses `hostagent/swap.py:preempt_for`; `pytest`/agent-level test the refuse-if-job path. -- [ ] **T482** [US5] Console switch surface (`ui/components/models/*`, `ui/components/serving/*`): the +- [X] **T482** [US5] Console switch surface (`ui/components/models/*`, `ui/components/serving/*`): the **promote action IS the switch** (Clarifications 2026-07-05 — no separate "set serving" control); when promoting a text-generation version would displace a resident serving model, gate it behind the 021 `ConfirmDialog` naming the holder. Show the **resident-vs-promoted** LLM delta (FR-269) and the base/adapter nature + lineage (FR-268); reflect a completed switch. Validate: `npm run build` green. -- [ ] **T483** [P] [US5] `ui/lib/gw-allowlist.ts` — add any new serving-control route the switch UI +- [X] **T483** [P] [US5] `ui/lib/gw-allowlist.ts` — add any new serving-control route the switch UI calls (prefer reusing existing entries; keep the delta minimal and equal to [contracts/agent-identity-and-allowlist.md](./contracts/agent-identity-and-allowlist.md)). - [ ] **T484** [US5] [HW] Validate US5 on the RTX 5070 Ti (quickstart §US5): the never-two-resident @@ -178,20 +182,23 @@ preempt of a serving holder, job never preempted (FR-257/258/259). ## Phase 8: Polish & cross-cutting -- [ ] **T485** [P] Full backend regression + boundary guards: `pytest` green across the resolver, +- [X] **T485** [P] Full backend regression + boundary guards: `pytest` green across the resolver, identity reporting, reload wiring, task-tag stamping, and backfill; confirm **no pre-existing test regressed** — especially the Principle II admission/swap tests (SC-147/149). Add two boundary checks: (a) **FR-273** — assert the feature adds **no new always-on resident process** and stays within the VRAM/RAM/disk budget (Principle III); (b) **FR-275** — assert the auto-promote/retraining policy path **cannot** switch the served LLM (served-LLM switching is operator-initiated only). -- [ ] **T486** [P] Console gate: `npm run lint` + `npm run build` green; allow-list conformance grep — +- [X] **T486** [P] Console gate: `npm run lint` + `npm run build` green; allow-list conformance grep — every gateway call in `ui/` resolves to a `gw-allowlist.ts` entry and the delta equals the contract (ideally empty — reuse preferred) (FR-272). -- [ ] **T487** [P] Docs: update the README serving section — the LLM is now registry-driven like the + *(build/type-check green; conformance grep clean with an EMPTY allow-list delta — the switch rides + the existing promote + serving/state routes. `next lint` remains unconfigured in-repo (no ESLint + config or dependency exists in `ui/` — pre-existing, unchanged by 022).)* +- [X] **T487** [P] Docs: update the README serving section — the LLM is now registry-driven like the other engines (promote → serve), base+adapter serving, and how to select the serving LLM; note the `LORA`/`MODEL` env knob (commit `c28ca97`) is now a fallback default, superseded by registry resolution. -- [ ] **T488** Constitution/agent-context: confirm **no amendment required** — Principle II's RULE is +- [X] **T488** Constitution/agent-context: confirm **no amendment required** — Principle II's RULE is unchanged (the switch reuses the existing one-tenant swap; this feature only chooses *which* model serves). Keep the managed `CLAUDE.md` marker on the 022 plan. - [ ] **T489** [HW] Full-loop drill on the RTX 5070 Ti: data → train → promote → **serve live** → diff --git a/tests/_llmregistry.py b/tests/_llmregistry.py new file mode 100644 index 0000000..04f220c --- /dev/null +++ b/tests/_llmregistry.py @@ -0,0 +1,132 @@ +"""In-memory stand-ins for the 022 serving-LLM seams (research R8 — offline, fake registry/store). + +`FakeRegistry` covers exactly the duck-typed MLflow client surface the resolution walk touches +(`get_model_version_by_alias` / `get_model_version` / `search_model_versions` / +`search_registered_models` / tag + alias writes), with the not-found shape `llmresolve` classifies +(`error_code == "RESOURCE_DOES_NOT_EXIST"`). `FakeLLMStore` mirrors `platformlib.store`'s +serving-LLM pointer surface, StoreError included, so `hostagent.serving_llm.active_model_name` and +the gateway pointer accessors run the real code paths with only the SQL/socket faked (the same +stance as tests/_agentstore.py). +""" +from platformlib import store as _store + +try: # the gateway registry code catches MlflowException; the agent code duck-types error_code — + # subclassing the real exception satisfies BOTH consumers of this one fake. + from mlflow.exceptions import MlflowException as _NotFoundBase +except Exception: # noqa: BLE001 — mlflow-less interpreter: agent-side tests still work + _NotFoundBase = Exception + + +class NotFoundError(_NotFoundBase): + def __init__(self, msg): + Exception.__init__(self, msg) # skip MlflowException.__init__ (proto error-code plumbing) + self.error_code = "RESOURCE_DOES_NOT_EXIST" + + +class FakeMV: + """A model-version row (the attribute surface mlflow's ModelVersion exposes to our code).""" + + def __init__(self, name, version, source, tags=None, run_id=None): + self.name, self.version = name, str(version) + self.source, self.tags, self.run_id = source, dict(tags or {}), run_id + self.status = "READY" + + +class FakeRM: + """A registered-model row for search_registered_models (name + aliases).""" + + def __init__(self, name, aliases): + self.name, self.aliases = name, dict(aliases or {}) + + +class FakeRegistry: + def __init__(self): + self.versions = [] # [FakeMV] + self.aliases = {} # model name -> the @serving version (str) + self.tag_writes = [] # (name, version, key, value) — asserts non-clobber/idempotency + + def add(self, name, version, source, tags=None, serving=False) -> FakeMV: + mv = FakeMV(name, version, source, tags) + self.versions.append(mv) + if serving: + self.aliases[name] = str(version) + return mv + + # -- the duck-typed client surface ------------------------------------------------------------ + def get_model_version_by_alias(self, name, alias): + v = self.aliases.get(name) + if alias != "serving" or v is None: + raise NotFoundError(f"RESOURCE_DOES_NOT_EXIST: no @{alias} for {name}") + return self.get_model_version(name, v) + + def get_model_version(self, name, version): + for mv in self.versions: + if mv.name == name and mv.version == str(version): + return mv + raise NotFoundError(f"RESOURCE_DOES_NOT_EXIST: {name} v{version}") + + def search_model_versions(self, flt): + if flt.startswith("name='") and flt.endswith("'"): + val = flt[len("name='"):-1].replace("''", "'") + return [mv for mv in self.versions if mv.name == val] + if flt.startswith("tags."): + key, _, val = flt[len("tags."):].partition("=") + val = val.strip("'").replace("''", "'") + return [mv for mv in self.versions if mv.tags.get(key) == val] + raise ValueError(f"fake registry cannot parse filter {flt!r}") + + def search_registered_models(self): + names = sorted({mv.name for mv in self.versions}) + return [FakeRM(n, {"serving": self.aliases[n]} if n in self.aliases else {}) + for n in names] + + def set_model_version_tag(self, name, version, key, value): + self.get_model_version(name, version).tags[key] = str(value) + self.tag_writes.append((name, str(version), key, str(value))) + + def set_registered_model_alias(self, name, alias, version): + self.aliases[name] = str(version) + + def create_registered_model(self, name): + return FakeRM(name, {}) + + def create_model_version(self, name, source, run_id=None, tags=None): + nxt = max([int(mv.version) for mv in self.versions if mv.name == name] or [0]) + 1 + return self.add(name, nxt, source, tags) + + +class FakeLLMStore: + """platformlib.store stand-in for the ActiveServingLLM pointer surface (T461).""" + + StoreError = _store.StoreError + + def __init__(self): + self.row = None # the singleton pointer row, or None (unset ⇒ default base) + self.fail = False # flip on to simulate a store outage + self.writes = 0 + + def connect(self, dsn=None, *, autocommit=True): + if self.fail: + raise self.StoreError("gateway DB unreachable (fake)") + return self + + def close(self): + pass + + def bootstrap(self, conn=None): + pass + + def get_serving_llm(self, conn): + if self.fail: + raise self.StoreError("gateway DB unreachable (fake)") + return dict(self.row) if self.row else None + + def set_serving_llm(self, conn, model_name, selected_at, selected_by): + if self.fail: + raise self.StoreError("gateway DB unreachable (fake)") + self.row = {"model_name": model_name, "selected_at": selected_at, + "selected_by": selected_by} + self.writes += 1 + + def clear_serving_llm(self, conn): + self.row = None diff --git a/tests/test_backfill_llm_tags.py b/tests/test_backfill_llm_tags.py new file mode 100644 index 0000000..d38f1c7 --- /dev/null +++ b/tests/test_backfill_llm_tags.py @@ -0,0 +1,153 @@ +"""022 T462/T477 — base-GGUF registration + the legacy task-tag backfill (offline, fake registry). + +Pins: `register_base_gguf` registers each PRESENT zoo base exactly once (re-run: nothing new; +absent file: skipped, never downloaded — R7) with the tags both resolution paths need +(kind=full-model + base_id). `backfill_llm_task_tags` stamps task/serving_engine on legacy +LLM-shaped versions (ops-bot-v1/v2), is idempotent (re-run is a no-op) and NON-CLOBBER (an +existing tag is never overwritten), and never touches non-LLM versions. +""" +import os +import sys + +import pytest + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if REPO not in sys.path: + sys.path.insert(0, REPO) +sys.path.insert(0, os.path.join(REPO, "scripts")) + +from _llmregistry import FakeRegistry # noqa: E402 +from backfill_llm_task_tags import backfill # noqa: E402 +from register_base_gguf import register_bases # noqa: E402 + + +def _quiet(*a, **kw): + pass + + +# -- T462: register_base_gguf ------------------------------------------------------------------------ + +def _bases(tmp_path, present=("b1.gguf",)): + for name in present: + (tmp_path / name).write_bytes(b"g" * 128) + return [{"name": "qwen-b1", "base_id": "Qwen/B1", "file": "b1.gguf"}, + {"name": "qwen-b2", "base_id": "Qwen/B2", "file": "b2.gguf"}] + + +def test_register_bases_registers_present_and_skips_absent(tmp_path): + reg = FakeRegistry() + report = register_bases(reg, bases=_bases(tmp_path), gguf_dir=str(tmp_path), log=_quiet) + assert [r["name"] for r in report["registered"]] == ["qwen-b1"] + assert [r["name"] for r in report["skipped"]] == ["qwen-b2"] # absent → skipped, no download + mv = reg.get_model_version("qwen-b1", "1") + assert mv.tags["kind"] == "full-model" and mv.tags["task"] == "text-generation" + assert mv.tags["base_id"] == "Qwen/B1" and mv.source == str(tmp_path / "b1.gguf") + + +def test_register_bases_is_idempotent(tmp_path): + reg = FakeRegistry() + bases = _bases(tmp_path) + register_bases(reg, bases=bases, gguf_dir=str(tmp_path), log=_quiet) + report = register_bases(reg, bases=bases, gguf_dir=str(tmp_path), log=_quiet) + assert report["registered"] == [] # re-run registers nothing new + assert len([mv for mv in reg.versions if mv.name == "qwen-b1"]) == 1 + + +# -- T477: backfill_llm_task_tags -------------------------------------------------------------------- + +def _legacy_registry(): + reg = FakeRegistry() + # the live gap: fine-tunes registered pre-022 — kind/format but NO task/serving_engine + reg.add("ops-bot-v1", 1, "s3://models/a1.gguf", + {"kind": "lora-adapter", "format": "gguf", "base_model": "Qwen/B1"}) + reg.add("ops-bot-v2", 1, "s3://models/a2.gguf", + {"kind": "lora-adapter", "format": "gguf", "base_model": "Qwen/B1"}) + # already-correct 022 registrations — must NOT be re-tagged + reg.add("ops-bot-v3", 1, "s3://models/a3.gguf", + {"kind": "lora-adapter", "format": "gguf", "task": "text-generation", + "serving_engine": "llama.cpp"}) + # non-LLM shape — never touched + reg.add("vision-net", 1, "s3://models/v.pt", {"kind": "vision-classifier"}) + return reg + + +def test_backfill_tags_legacy_versions(): + reg = _legacy_registry() + report = backfill(reg, log=_quiet) + tagged = {r["name"] for r in report["tagged"]} + assert tagged == {"ops-bot-v1", "ops-bot-v2"} + for name in ("ops-bot-v1", "ops-bot-v2"): + tags = reg.get_model_version(name, "1").tags + assert tags["task"] == "text-generation" and tags["serving_engine"] == "llama.cpp" + assert "task" not in reg.get_model_version("vision-net", "1").tags + + +def test_backfill_is_idempotent_and_non_clobber(): + reg = _legacy_registry() + # a version with a DIFFERENT existing task tag must never be clobbered + reg.add("weird", 1, "s3://models/w.gguf", + {"kind": "lora-adapter", "task": "custom-task"}) + backfill(reg, log=_quiet) + writes_after_first = len(reg.tag_writes) + report = backfill(reg, log=_quiet) + assert report["tagged"] == [] # re-run is a no-op + assert len(reg.tag_writes) == writes_after_first # zero additional writes + assert reg.get_model_version("weird", "1").tags["task"] == "custom-task" # non-clobber + # 'weird' got only the missing serving_engine on the first run, task untouched + assert reg.get_model_version("weird", "1").tags["serving_engine"] == "llama.cpp" + + +# -- T476: the fine-tune flow stamps task descriptors at registration -------------------------------- + +def test_finetune_registration_stamps_task_descriptors(monkeypatch, tmp_path): + import types + + sys.path.insert(0, os.path.join(REPO, "training", "flows")) + import finetune + + class FakeS3: + def put_object(self, **kw): + pass + + captured = {} + + class FakeClient: + def __init__(self, tracking_uri=None): + pass + + def create_registered_model(self, name): + pass + + def create_model_version(self, name, source, run_id, tags): + captured.update(name=name, source=source, tags=tags) + + class MV: + version = "7" + + return MV() + + # register_version imports `from mlflow.tracking import MlflowClient` at CALL time — stub the + # sys.modules seam directly so this is independent of whatever mlflow state (real module or a + # prior test's stub) the suite left behind. + fake_tracking = types.ModuleType("mlflow.tracking") + fake_tracking.MlflowClient = FakeClient + fake_exceptions = types.ModuleType("mlflow.exceptions") + fake_exceptions.MlflowException = type("MlflowException", (Exception,), {}) + monkeypatch.setitem(sys.modules, "mlflow.tracking", fake_tracking) + monkeypatch.setitem(sys.modules, "mlflow.exceptions", fake_exceptions) + monkeypatch.setattr(finetune, "_s3", lambda: FakeS3()) + gguf = tmp_path / "a.gguf" + gguf.write_bytes(b"g") + register = getattr(finetune.register_version, "fn", finetune.register_version) + out = register("ops-bot", str(gguf), "run123", "Qwen/B1", "ops-qa", "v3") + # FR-266: a NEW fine-tune registers as a first-class text-generation serving target + assert captured["tags"]["task"] == "text-generation" + assert captured["tags"]["serving_engine"] == "llama.cpp" + assert captured["tags"]["kind"] == "lora-adapter" + assert captured["tags"]["base_model"] == "Qwen/B1" + assert captured["tags"]["dataset_name"] == "ops-qa" + assert out["version"] == "7" + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-q"])) diff --git a/tests/test_llama_registry_binding.py b/tests/test_llama_registry_binding.py new file mode 100644 index 0000000..252b68b --- /dev/null +++ b/tests/test_llama_registry_binding.py @@ -0,0 +1,187 @@ +"""022 T465/T469 — the llama adapter's registry binding + honest identity (offline, GPU-free). + +Pins: `rebind()` binds base/adapter/alias/version from the resolver before spawn (`-m base +--lora adapter --alias `); the env knobs are the FALLBACK only (pointer unset, or +resolution infra unreachable — surfaced as `binding` in health, never silent); an unresolvable +target makes the engine unavailable (fail loud, FR-265 — the swap target-probe refuses); the +health payload carries the loaded model_name + registry_version (the agent is the identity +source of truth, FR-260); the forward/stream `model` field names the registry model (FR-262). +""" +import json +import os +import sys +import urllib.request + +import pytest + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if REPO not in sys.path: + sys.path.insert(0, REPO) + +from hostagent import serving_llm # noqa: E402 +from hostagent.adapters.llama import LlamaAdapter # noqa: E402 + + +def _files(tmp_path): + bin_ = tmp_path / "llama-server" + bin_.write_text("#!/bin/sh\n") + os.chmod(bin_, 0o755) + base = tmp_path / "base.gguf" + base.write_bytes(b"b" * (2 * 1024 * 1024)) + adapter = tmp_path / "adapter.gguf" + adapter.write_bytes(b"a" * 1024) + return str(bin_), str(base), str(adapter) + + +def _adapter(tmp_path, monkeypatch, resolver, env_model=None): + bin_, base, adapter = _files(tmp_path) + monkeypatch.setenv("LLAMA_BIN", bin_) + monkeypatch.setenv("MODEL", env_model or base) + monkeypatch.delenv("LORA", raising=False) + a = LlamaAdapter(resolver=resolver) + return a, base, adapter + + +def _target(base, adapter=None, name="ops-bot", version="3"): + return {"model_name": name, "version": version, + "kind": "lora-adapter" if adapter else "full-model", + "base_gguf": base, "adapter_gguf": adapter, + "base": {"name": "qwen-base", "version": "1", "source": base} if adapter else None} + + +class FakePopen: + def __init__(self, cmd): + self.cmd, self.pid = cmd, 4242 + + def poll(self): + return None + + +def test_resolved_adapter_binds_spawn_args_and_identity(tmp_path, monkeypatch): + a, base, adapter = _adapter(tmp_path, monkeypatch, + resolver=lambda: _target(None, None)) + a._resolver = lambda: _target(base, adapter) + captured = {} + monkeypatch.setattr("subprocess.Popen", lambda cmd, **kw: captured.update(cmd=cmd) or + FakePopen(cmd), raising=False) + import hostagent.adapters.llama as llama_mod + monkeypatch.setattr(llama_mod.subprocess, "Popen", lambda cmd, **kw: FakePopen(cmd)) + assert a.available() == (True, None) + child = a.spawn() + cmd = child.cmd + assert cmd[cmd.index("-m") + 1] == base + assert cmd[cmd.index("--lora") + 1] == adapter + assert cmd[cmd.index("--alias") + 1] == "ops-bot" + assert a.bound_identity() == ("ops-bot", "3") + assert a.loaded_identity() == ("ops-bot", "3") + + +def test_full_model_spawn_has_no_lora_flag(tmp_path, monkeypatch): + a, base, _ = _adapter(tmp_path, monkeypatch, resolver=None) + a._resolver = lambda: _target(base, None, name="qwen-base", version="1") + import hostagent.adapters.llama as llama_mod + monkeypatch.setattr(llama_mod.subprocess, "Popen", lambda cmd, **kw: FakePopen(cmd)) + a.rebind(force=True) + child = a.spawn() + assert "--lora" not in child.cmd and child.cmd[child.cmd.index("-m") + 1] == base + + +def test_unset_pointer_falls_back_to_env_defaults(tmp_path, monkeypatch): + a, base, _ = _adapter(tmp_path, monkeypatch, resolver=lambda: None) + a.rebind(force=True) + assert a.model == base and a.lora is None + assert a.alias == a._env_alias and a.registry_version is None + h = a.health(resident=False) + assert "binding" not in h # a clean default is not a degradation + + +def test_unreachable_resolution_falls_back_and_surfaces_note(tmp_path, monkeypatch): + def resolver(): + raise serving_llm.ResolutionUnavailable("store down") + + a, base, _ = _adapter(tmp_path, monkeypatch, resolver=resolver) + assert a.available() == (True, None) # env default still serves (today's behavior) + h = a.health(resident=False) + assert "binding" in h and "store down" in h["binding"] + assert h["model_name"] == a._env_alias and h["registry_version"] is None + + +def test_unresolvable_target_makes_engine_unavailable(tmp_path, monkeypatch): + def resolver(): + raise serving_llm.ResolutionError("base 'x' is not registered") + + a, _, _ = _adapter(tmp_path, monkeypatch, resolver=resolver) + ok, reason = a.available() + assert not ok and "unresolvable" in reason and "base 'x'" in reason + + +def test_missing_bound_adapter_file_is_unavailable(tmp_path, monkeypatch): + a, base, _ = _adapter(tmp_path, monkeypatch, resolver=None) + a._resolver = lambda: _target(base, str(tmp_path / "gone-adapter.gguf")) + ok, reason = a.available() + assert not ok and "LoRA adapter" in reason + + +def test_rebind_is_ttl_cached_and_force_busts_it(tmp_path, monkeypatch): + calls = [] + + def resolver(): + calls.append(1) + return None + + a, _, _ = _adapter(tmp_path, monkeypatch, resolver=resolver) + a.available() + a.available() + assert len(calls) == 1 # second available() within the TTL — no new resolve + a.rebind(force=True) + assert len(calls) == 2 # the reload verb busts the cache + + +def test_health_reports_loaded_identity_when_resident(tmp_path, monkeypatch): + a, base, adapter = _adapter(tmp_path, monkeypatch, resolver=None) + a._resolver = lambda: _target(base, adapter) + import hostagent.adapters.llama as llama_mod + monkeypatch.setattr(llama_mod.subprocess, "Popen", lambda cmd, **kw: FakePopen(cmd)) + a.rebind(force=True) + a.spawn() + h = a.health(resident=True) + assert h["model"] == "ops-bot" and h["model_name"] == "ops-bot" + assert h["registry_version"] == "3" + assert h["base"] == os.path.basename(base) and h["adapter"] == os.path.basename(adapter) + + +def test_forward_and_stream_echo_the_registry_identity(tmp_path, monkeypatch): + a, base, adapter = _adapter(tmp_path, monkeypatch, resolver=None) + a._resolver = lambda: _target(base, adapter) + import hostagent.adapters.llama as llama_mod + monkeypatch.setattr(llama_mod.subprocess, "Popen", lambda cmd, **kw: FakePopen(cmd)) + a.rebind(force=True) + a.spawn() + a._port = 9999 + + body = json.dumps({"choices": [{"message": {"content": "hi"}}], "usage": {}}).encode() + + class _Resp: + status = 200 + + def read(self, *a): + return body + + def __iter__(self): + return iter([b'data: [DONE]\n']) + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + monkeypatch.setattr(urllib.request, "urlopen", lambda req, timeout=None: _Resp()) + out = a.forward("infer", {"prompt": "x"}, 0.0) + assert out["model"] == "ops-bot" # FR-262: the response names the registry model + frames = list(a.stream("infer", {"prompt": "x"}, 0.0)) + assert b'"model": "ops-bot"' in frames[0] and b'"model": "ops-bot"' in frames[-1] + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-q"])) diff --git a/tests/test_llm_serving_boundaries.py b/tests/test_llm_serving_boundaries.py new file mode 100644 index 0000000..ab42944 --- /dev/null +++ b/tests/test_llm_serving_boundaries.py @@ -0,0 +1,99 @@ +"""022 T485 — the two cross-cutting boundary guards (FR-273 footprint, FR-275 automation). + + - **FR-273 (Principle III)**: registry-driven LLM serving adds NO new always-on resident + process — the compose service set and the agent's engine registry are unchanged, resolution + is cold-load-only (importing the resolver spawns no thread), and the only new persisted state + is the one-row pointer table. + - **FR-275 (operator-only switch)**: the retraining/auto-promote policy path MAY gate and + register a candidate but can NEVER switch the served LLM. Structurally: `registry.promote` + (the choke-point the scheduler + suggestion-accept call) touches neither the pointer nor the + reload; only the operator's `POST /models/{name}/promote` route wires the go-live half. +""" +import os +import re +import sys +import threading + +import pytest + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +for _p in (REPO, os.path.join(REPO, "gateway")): + if _p not in sys.path: + sys.path.insert(0, _p) + +from _llmregistry import FakeRegistry # noqa: E402 +from app import registry # noqa: E402 + +#: The resident compose control plane as of 020/021 — 022 must not grow it (FR-273). +EXPECTED_SERVICES = {"postgres", "garage", "garage-init", "mlflow", + "gateway", "prometheus", "grafana"} + + +def _compose_services(): + text = open(os.path.join(REPO, "docker-compose.yml"), encoding="utf-8").read() + body = text[text.index("\nservices:"):] + if "\nvolumes:" in body: + body = body[:body.index("\nvolumes:")] + return set(re.findall(r"^ (\w[\w-]*):\s*$", body, flags=re.M)) + + +# -- FR-273: no new resident process ---------------------------------------------------------------- + +def test_compose_service_set_is_unchanged(): + assert _compose_services() == EXPECTED_SERVICES + + +def test_agent_engine_registry_is_unchanged(): + from platformlib.topology import ENGINES + assert set(ENGINES) == {"llm", "asr", "vision", "embed", "tabular"} + + +def test_resolver_import_spawns_no_thread(): + before = threading.active_count() + import hostagent.serving_llm # noqa: F401 — cold-load-only: no daemon, no watcher + import platformlib.llmresolve # noqa: F401 + assert threading.active_count() == before + + +def test_only_new_persisted_state_is_the_pointer_row(): + from platformlib import store + # exactly one table added vs the 018 set — the singleton pointer (data-model.md) + assert set(store.TABLES) - {"meta", "predictions", "labels", "capture_index", "jobs", + "policies", "suggestions"} == {"serving_llm"} + + +# -- FR-275: the policy path cannot switch the served LLM -------------------------------------------- + +def test_registry_promote_never_touches_pointer_or_reload(monkeypatch, tmp_path): + # The scheduler's auto-promote and the suggestion accept BOTH call registry.promote directly + # (tests below pin that) — so proving promote() itself never goes live proves the boundary. + reg = FakeRegistry() + reg.add("ops-bot", 2, "s3://models/a.gguf", + {"kind": "lora-adapter", "task": "text-generation", "base_model": "qwen-base"}) + monkeypatch.setattr(registry, "_client", lambda: reg) + from app import evaluation + monkeypatch.setattr(evaluation, "gate", + lambda name, version, override=False, client=None: + {"verdict": "pass", "reason": "test"}) + touched = [] + monkeypatch.setattr(registry, "set_serving_llm", + lambda *a, **kw: touched.append("pointer")) + from app import serving + monkeypatch.setattr(serving, "request_llm_reload", + lambda *a, **kw: touched.append("reload")) + res = registry.promote("ops-bot", "2") + assert res["promoted"] is True and reg.aliases["ops-bot"] == "2" # the alias DID move + assert touched == [] # …but nothing went live: no pointer write, no agent reload (FR-275) + + +def test_policy_and_scheduler_paths_reference_no_go_live_surface(): + # Belt-and-braces source pin: the auto-promote path (scheduler) and the suggestion accept + # (routers/policies) must keep calling registry.promote — never the go-live half. + for rel in ("gateway/app/scheduler.py", "gateway/app/routers/policies.py"): + src = open(os.path.join(REPO, rel), encoding="utf-8").read() + assert "set_serving_llm" not in src, rel + assert "request_llm_reload" not in src, rel + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-q"])) diff --git a/tests/test_serving_llm_gateway.py b/tests/test_serving_llm_gateway.py new file mode 100644 index 0000000..3afd8cf --- /dev/null +++ b/tests/test_serving_llm_gateway.py @@ -0,0 +1,249 @@ +"""022 T464/T467/T470/T471/T474/T478/T479 — the gateway half of registry-driven LLM serving. + +Offline (research R8): the MLflow client and the pointer store are the fakes in +tests/_llmregistry.py. Pins: + + - the ActiveServingLLM pointer accessors (T464) — round-trip, unset default, fail-loud write; + - `llm_target_info` (T474): an adapter whose base does not resolve is refused BEFORE the gate, + so the alias and the currently-served LLM stay unchanged (FR-265); + - the promote route wiring (T467): promote = go live — pointer write + agent reload, operator + `preempt` passed through; a gate-blocked promote moves nothing; + - `resolve_serving_target`/`list_tasks` kind fallback (T478, FR-267) + kind/lineage exposure and + the FR-276 active-pointer dedup (T479): two promoted LLM models → exactly ONE live target; + - honest identity consumption (T470/T471, FR-260/261/262): serving state + the /infer response + + the logged prediction all carry the AGENT-reported identity, degrading to `unknown`. +""" +import asyncio +import os +import sys + +import pytest + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +for _p in (REPO, os.path.join(REPO, "gateway")): + if _p not in sys.path: + sys.path.insert(0, _p) + +from _llmregistry import FakeLLMStore, FakeRegistry # noqa: E402 +from app import registry, serving # noqa: E402 +from app.routers import infer as infer_router # noqa: E402 +from app.routers import models as models_router # noqa: E402 +from fastapi import HTTPException # noqa: E402 + + +@pytest.fixture +def fake_reg(monkeypatch): + reg = FakeRegistry() + monkeypatch.setattr(registry, "_client", lambda: reg) + return reg + + +@pytest.fixture +def fake_store(monkeypatch): + s = FakeLLMStore() + monkeypatch.setattr(registry, "_store", lambda: s) + return s + + +# -- T464: pointer accessors ------------------------------------------------------------------------- + +def test_pointer_round_trip_and_default(fake_store): + assert registry.get_serving_llm() is None + assert registry.active_serving_llm_name() == registry.DEFAULT_LLM # unset ⇒ default base + registry.set_serving_llm("ops-bot", actor="operator") + assert registry.get_serving_llm()["model_name"] == "ops-bot" + assert registry.active_serving_llm_name() == "ops-bot" + + +def test_pointer_write_fails_loud_read_fails_soft(fake_store): + fake_store.fail = True + assert registry.get_serving_llm() is None # read degrades to 'unset' (default base) + with pytest.raises(registry.RegistryError, match="pointer write failed"): + registry.set_serving_llm("ops-bot") + + +# -- T474: pre-promotion resolution check ------------------------------------------------------------ + +def test_target_info_none_for_non_llm(fake_reg): + fake_reg.add("vision-net", 1, "s3://models/v.pt", {"task": "image-classification"}) + assert registry.llm_target_info("vision-net", "1") is None + + +def test_target_info_resolves_adapter_base(fake_reg): + fake_reg.add("qwen-base", 1, "/zoo/base.gguf", {"kind": "full-model"}) + fake_reg.add("ops-bot", 2, "s3://models/a.gguf", + {"kind": "lora-adapter", "base_model": "qwen-base", "task": "text-generation"}) + info = registry.llm_target_info("ops-bot", "2") + assert info["kind"] == "lora-adapter" and info["error"] is None + assert info["base"]["name"] == "qwen-base" and info["base"]["source"] == "/zoo/base.gguf" + + +def test_target_info_unresolvable_base_carries_error(fake_reg): + fake_reg.add("ops-bot", 1, "s3://models/a.gguf", + {"kind": "lora-adapter", "base_model": "nowhere", "task": "text-generation"}) + info = registry.llm_target_info("ops-bot", "1") + assert info["error"] and "register the local base" in info["error"] + + +def test_target_info_recognizes_legacy_untagged_adapter(fake_reg): + # FR-267: kind/format only, no task tag — still a text-generation target (T478 inference). + fake_reg.add("qwen-base", 1, "/zoo/base.gguf", {"kind": "full-model"}) + fake_reg.add("ops-bot-v1", 1, "s3://models/a.gguf", + {"kind": "lora-adapter", "base_model": "qwen-base"}) + info = registry.llm_target_info("ops-bot-v1", "1") + assert info is not None and info["task"] == "text-generation" and info["error"] is None + + +# -- T467: promote = go live ------------------------------------------------------------------------- + +def _promote_env(monkeypatch, *, target, promoted=True): + calls = {"promote": [], "pointer": [], "reload": []} + monkeypatch.setattr(registry, "list_versions", + lambda name: [{"version": "2"}]) + monkeypatch.setattr(registry, "llm_target_info", lambda n, v: target) + monkeypatch.setattr(registry, "promote", + lambda n, v, override=False: + calls["promote"].append((n, v)) or + {"name": n, "serving_version": v, "promoted": promoted, + "verdict": {"verdict": "pass"}}) + monkeypatch.setattr(registry, "set_serving_llm", + lambda n, actor="operator": calls["pointer"].append(n) or + {"model_name": n}) + monkeypatch.setattr(serving, "request_llm_reload", + lambda preempt=False: calls["reload"].append(preempt) or + {"status": "reloaded"}) + return calls + + +def test_promote_llm_writes_pointer_and_requests_reload(monkeypatch): + target = {"task": "text-generation", "kind": "lora-adapter", + "base": {"name": "qwen-base", "version": "1", "source": "/zoo/b.gguf"}, + "error": None} + calls = _promote_env(monkeypatch, target=target) + res = models_router.promote("ops-bot", models_router.PromoteRequest(version="2", preempt=True)) + assert calls["promote"] == [("ops-bot", "2")] + assert calls["pointer"] == ["ops-bot"] # promote IS the go-live action + assert calls["reload"] == [True] # operator confirm passed through (FR-258) + assert res["serving_llm"]["reload"]["status"] == "reloaded" + + +def test_promote_refuses_unresolvable_llm_before_the_gate(monkeypatch): + target = {"task": "text-generation", "kind": "lora-adapter", "base": None, + "error": "base 'x' is not a registered full-model version"} + calls = _promote_env(monkeypatch, target=target) + with pytest.raises(HTTPException) as exc: + models_router.promote("ops-bot", models_router.PromoteRequest(version="2")) + assert exc.value.status_code == 409 and "refused" in exc.value.detail + assert calls["promote"] == [] and calls["pointer"] == [] # alias + served LLM unchanged + + +def test_promote_non_llm_touches_no_pointer(monkeypatch): + calls = _promote_env(monkeypatch, target=None) + res = models_router.promote("vision-net", models_router.PromoteRequest(version="2")) + assert calls["promote"] and calls["pointer"] == [] and calls["reload"] == [] + assert "serving_llm" not in res + + +def test_gate_blocked_promote_moves_nothing(monkeypatch): + target = {"task": "text-generation", "kind": "full-model", "base": None, "error": None} + calls = _promote_env(monkeypatch, target=target, promoted=False) + models_router.promote("qwen", models_router.PromoteRequest(version="2")) + assert calls["pointer"] == [] and calls["reload"] == [] # blocked verdict → not live + + +# -- T478/T479: task surfaces ------------------------------------------------------------------------ + +def test_list_tasks_infers_task_and_exposes_kind_lineage(fake_reg, fake_store): + fake_reg.add("ops-bot", 1, "s3://models/a.gguf", + {"kind": "lora-adapter", "base_model": "qwen-base", + "dataset_name": "ops-qa", "dataset_version": "v3"}, serving=True) + tasks = registry.list_tasks() + entry = next(t for t in tasks if t["model"] == "ops-bot") + assert entry["task"] == "text-generation" # inferred, not null → a real panel (FR-267) + assert entry["kind"] == "lora-adapter" + assert entry["lineage"]["base_model"] == "qwen-base" + assert entry["lineage"]["dataset_name"] == "ops-qa" + + +def test_list_tasks_filters_to_the_active_pointer(fake_reg, fake_store): + # FR-276: a promote moves only its own model's alias, so BOTH models keep one — only the + # active-pointer model may appear as the live text-generation target (no stale duplicate). + fake_reg.add("qwen", 5, "/zoo/base.gguf", + {"kind": "full-model", "task": "text-generation"}, serving=True) + fake_reg.add("ops-bot", 2, "s3://models/a.gguf", + {"kind": "lora-adapter", "base_model": "qwen", "task": "text-generation"}, + serving=True) + def live_llms(): + return [t["model"] for t in registry.list_tasks() if t["task"] == "text-generation"] + + fake_store.set_serving_llm(None, "ops-bot", 1000.0, "operator") + assert live_llms() == ["ops-bot"] + # switch back → the other model is the single live target again, no stale duplicate + fake_store.set_serving_llm(None, "qwen", 1001.0, "operator") + assert live_llms() == ["qwen"] + + +def test_resolve_serving_target_falls_back_to_kind(fake_reg): + # A legacy promoted LLM with NO task tag is invisible to the tag search — the kind fallback + # still routes it (T478). + fake_reg.add("ops-bot-v1", 1, "s3://models/a.gguf", + {"kind": "lora-adapter", "base_model": "qwen-base"}, serving=True) + target = registry.resolve_serving_target("text-generation") + assert target is not None and target["name"] == "ops-bot-v1" + assert target["serving_engine"] == "llama.cpp" + + +# -- T470/T471: honest identity consumption ---------------------------------------------------------- + +def test_identity_from_health_maps_agent_fields(): + ident = serving._identity_from_health( + {"model_name": "ops-bot", "registry_version": "3", + "base": "base.gguf", "adapter": "a.gguf", "model": "ops-bot"}) + assert ident == {"serving_model": "ops-bot", "serving_version": "3", + "base": "base.gguf", "adapter": "a.gguf"} + + +def test_identity_degrades_to_unknown_never_a_config_guess(): + assert serving._identity_from_health({})["serving_model"] == "unknown" + assert serving._UNKNOWN_IDENTITY["serving_version"] is None + + +def test_infer_attributes_response_and_prediction_to_agent_identity(monkeypatch): + logged = [] + + async def fake_health(): + return True + + async def fake_run(prompt, max_tokens, temperature, *, preempt=False): + return {"text": "hi", "load_ms": 0.0, "infer_ms": 1.0, "model": "ops-bot", "usage": {}} + + async def fake_identity(): + return {"serving_model": "ops-bot", "serving_version": "3", + "base": "base.gguf", "adapter": "a.gguf"} + + monkeypatch.setattr(infer_router, "health", fake_health) + monkeypatch.setattr(infer_router, "run_inference", fake_run) + monkeypatch.setattr(infer_router, "llm_identity", fake_identity) + monkeypatch.setattr(infer_router.quality, "log_prediction", + lambda model, version, modality, *a, **kw: + logged.append((model, version, modality)) or "pid-1") + monkeypatch.setattr(infer_router.quality, "capture_input", lambda *a, **kw: None) + monkeypatch.setattr(infer_router.tracing, "emit", lambda **kw: None) + + res = asyncio.run(infer_router.infer(infer_router.InferRequest(prompt="x"))) + # FR-262: response identity == logged identity == the agent-reported served identity + assert res["registry_model"] == "ops-bot" and res["registry_version"] == "3" + assert logged == [("ops-bot", "3", "text-generation")] + + +def test_stream_log_uses_agent_identity_not_config(): + # Source pin (read from disk — import-order-independent): the detached _log resolves the + # served identity from the agent; the old SERVING_MODEL config-guess is gone. + path = os.path.join(REPO, "gateway", "app", "routers", "stream.py") + src = open(path, encoding="utf-8").read() + assert "llm_identity" in src + assert "log_prediction(serving.SERVING_MODEL" not in src + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-q"])) diff --git a/tests/test_serving_llm_reload.py b/tests/test_serving_llm_reload.py new file mode 100644 index 0000000..055033e --- /dev/null +++ b/tests/test_serving_llm_reload.py @@ -0,0 +1,153 @@ +"""022 T466/T481 — reload-on-select under the single-GPU lease (offline, GPU-free). + +Pins the two-case switch (contracts/serving-resolution.md, spec review PR #64 §1): +cross-tenant displacement reuses the transactional swap; the SAME-TENANT model switch (llm +resident with a different model — where preempt_for is a satisfied no-op) force-reloads +(unload → ensure_loaded) so llama-server actually re-spawns with the new artifact. Also pins: +idempotent no-op for the already-resident model+version (FR-256); job holders never preempted +(FR-259); a GPU batch's llm holder never reloaded under it (FR-155); displacement demands the +operator confirm (`preempt=true`, FR-258); the target is probed BEFORE any unload so a bad +artifact never takes down a working holder (FR-257); strictly sequential — never two children +alive across the switch (SC-147). +""" +import os +import sys + +import pytest + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if REPO not in sys.path: + sys.path.insert(0, REPO) + +from hostagent import admission as adm # noqa: E402 +from hostagent import lifecycle, swap # noqa: E402 +from test_agent_lifecycle import FakeEngine # noqa: E402 — the shared fake adapter + + +class FakeLlm(FakeEngine): + """FakeEngine + the 022 binding surface the reload verb drives.""" + + def __init__(self, est=1.0): + super().__init__("llm", True, est) + self.bound = ("model-a", "1") # what the resolver currently points at + self.loaded = None # captured at spawn (the resident identity) + self.rebinds = [] + + def rebind(self, force=False): + self.rebinds.append(force) + + def bound_identity(self): + return self.bound + + def loaded_identity(self): + return self.loaded + + def spawn(self): + self.loaded = self.bound + return super().spawn() + + +def _manager(extra=()): + a = adm.Admission(vram_budget_gb=12.0, + gpu=adm.GpuReader(ttl_s=1000.0, read_fn=lambda: 10.0)) + llm = FakeLlm() + engines = [llm, *extra] + runtimes = {e.engine_id: lifecycle.EngineRuntime(e, a, sleep=lambda s: None) + for e in engines} + return lifecycle.EngineManager(a, runtimes), a, llm + + +def test_idle_gpu_loads_the_selected_llm_immediately(): + mgr, a, llm = _manager() + res = swap.reload_serving_llm(mgr) + assert res["status"] == "loaded" and res["model_name"] == "model-a" + assert a.holder()["tenant"] == "llm" and llm.loaded == ("model-a", "1") + assert llm.rebinds == [True] # the promote's reload always resolves FRESH (TTL busted) + + +def test_same_model_resident_is_an_idempotent_noop(): + mgr, a, llm = _manager() + mgr.runtimes["llm"].ensure_loaded() + res = swap.reload_serving_llm(mgr, preempt=True) + assert res["status"] == "noop" and len(llm.spawned) == 1 # no gratuitous reload (FR-256) + + +def test_same_tenant_model_switch_force_reloads(): + # THE case preempt_for silently no-ops on (PR #64 §1): llm resident with model-a, promote + # model-b — the child must actually re-spawn with the new artifact. + mgr, a, llm = _manager() + mgr.runtimes["llm"].ensure_loaded() + first_child = llm.spawned[0] + llm.bound = ("model-b", "2") + res = swap.reload_serving_llm(mgr, preempt=True) + assert res["status"] == "reloaded" and res["model_name"] == "model-b" + assert len(llm.spawned) == 2 and llm.loaded == ("model-b", "2") + assert not first_child.alive # old child torn down BEFORE the new spawn + assert a.holder()["tenant"] == "llm" + + +def test_switch_requires_operator_confirm_when_displacing(): + mgr, a, llm = _manager() + mgr.runtimes["llm"].ensure_loaded() + llm.bound = ("model-b", "2") + with pytest.raises(swap.PreemptRefused, match="confirmation required"): + swap.reload_serving_llm(mgr) # no preempt=true → FR-258 refusal + assert llm.spawned[0].alive and llm.loaded == ("model-a", "1") # nothing displaced + + +def test_cross_tenant_serving_holder_swaps_with_confirm(): + vision = FakeEngine("vision") + mgr, a, llm = _manager(extra=[vision]) + mgr.runtimes["vision"].ensure_loaded() + with pytest.raises(swap.PreemptRefused, match="confirmation required"): + swap.reload_serving_llm(mgr) + res = swap.reload_serving_llm(mgr, preempt=True) + assert res["status"] == "swapped" and res["evicted"] == "vision" + assert a.holder()["tenant"] == "llm" and not vision.spawned[0].alive + + +def test_job_holder_is_never_preempted(): + mgr, a, llm = _manager() + a.acquire("training", "job", est_gb=6.0) # a running fine-tune owns the slot + with pytest.raises(swap.PreemptRefused, match="never preempted"): + swap.reload_serving_llm(mgr, preempt=True) + assert a.holder()["tenant"] == "training" # untouched; the switch is deferred (FR-259) + + +def test_batch_driven_llm_holder_is_not_reloaded_under_the_batch(): + mgr, a, llm = _manager() + mgr.runtimes["llm"].ensure_loaded() + llm.bound = ("model-b", "2") + with pytest.raises(swap.PreemptRefused, match="FR-155"): + swap.reload_serving_llm(mgr, preempt=True, batch_active_fn=lambda: True) + assert llm.spawned[0].alive and llm.loaded == ("model-a", "1") + + +def test_unloadable_target_refuses_before_any_eviction(): + mgr, a, llm = _manager() + mgr.runtimes["llm"].ensure_loaded() + llm.bound = ("model-b", "2") + llm.available_state = (False, "base GGUF not found") + with pytest.raises(swap.PreemptRefused, match="not loadable"): + swap.reload_serving_llm(mgr, preempt=True) + assert llm.spawned[0].alive and llm.loaded == ("model-a", "1") # holder untouched (FR-257) + + +def test_never_two_children_alive_across_the_switch(): + mgr, a, llm = _manager() + mgr.runtimes["llm"].ensure_loaded() + llm.bound = ("model-b", "2") + alive_counts = [] + orig_spawn = llm.spawn + + def counting_spawn(): + alive_counts.append(sum(1 for c in llm.spawned if c.alive)) + return orig_spawn() + + llm.spawn = counting_spawn + swap.reload_serving_llm(mgr, preempt=True) + assert alive_counts == [0] # at the moment of the new spawn, the old child was already dead + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-q"])) diff --git a/tests/test_serving_llm_resolver.py b/tests/test_serving_llm_resolver.py new file mode 100644 index 0000000..7188e34 --- /dev/null +++ b/tests/test_serving_llm_resolver.py @@ -0,0 +1,193 @@ +"""022 T461/T463/T473 — the ActiveServingLLM pointer + the host-agent cold-load resolver. + +Offline, GPU-free (research R8): the pointer store and the registry are the fakes in +tests/_llmregistry.py; artifacts are tmp files. Pins the serving-resolution contract: +full-model → `-m source`; lora-adapter → base resolved from lineage in ONE hop (registered name +or `base_id` tag — never an adapter chain); unset pointer → None (the configured default base); +invalid target → ResolutionError (the promote/select refusal, FR-265); unreachable infra → +ResolutionUnavailable (the env-fallback seam, never a silent guess). +""" +import os +import sys + +import pytest + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if REPO not in sys.path: + sys.path.insert(0, REPO) + +from _llmregistry import FakeLLMStore, FakeRegistry # noqa: E402 +from hostagent import serving_llm # noqa: E402 +from platformlib import llmresolve, store # noqa: E402 + + +def _pointed_store(name="ops-bot"): + s = FakeLLMStore() + s.set_serving_llm(None, name, 1000.0, "operator") + return s + + +def _gguf(tmp_path, name): + p = tmp_path / name + p.write_bytes(b"g" * 1024) + return str(p) + + +# -- T461: the pointer table + accessors ------------------------------------------------------------ + +def test_ddl_covers_the_serving_llm_pointer_table(): + assert "CREATE TABLE IF NOT EXISTS serving_llm " in store.DDL + assert "serving_llm" in store.TABLES + # additive-only — the schema version does not bump for a new IF NOT EXISTS table + assert store.SCHEMA_VERSION == 1 + + +def test_pointer_round_trip_and_unset_default(): + s = FakeLLMStore() + assert serving_llm.active_model_name(store=s) is None # fresh store → default base + s.set_serving_llm(None, "ops-bot", 1000.0, "operator") + assert serving_llm.active_model_name(store=s) == "ops-bot" + s.clear_serving_llm(None) + assert serving_llm.active_model_name(store=s) is None + + +def test_pointer_store_outage_is_unavailable_not_a_guess(): + s = FakeLLMStore() + s.fail = True + with pytest.raises(serving_llm.ResolutionUnavailable): + serving_llm.active_model_name(store=s) + + +# -- T463: full-model resolution -------------------------------------------------------------------- + +def test_full_model_resolves_source_as_base(tmp_path): + base = _gguf(tmp_path, "base.gguf") + reg = FakeRegistry() + reg.add("qwen-base", 1, base, {"kind": "full-model", "task": "text-generation"}, serving=True) + out = serving_llm.resolve(store=_pointed_store("qwen-base"), client=reg) + assert out["model_name"] == "qwen-base" and out["version"] == "1" + assert out["base_gguf"] == base and out["adapter_gguf"] is None + assert out["kind"] == "full-model" and out["base"] is None + + +def test_unset_pointer_resolves_to_none(): + assert serving_llm.resolve(store=FakeLLMStore(), client=FakeRegistry()) is None + + +def test_no_serving_alias_is_a_resolution_error(): + reg = FakeRegistry() + reg.add("ops-bot", 1, "s3://models/a.gguf", {"kind": "lora-adapter"}) # nothing promoted + with pytest.raises(serving_llm.ResolutionError, match="no @serving"): + serving_llm.resolve(store=_pointed_store("ops-bot"), client=reg) + + +def test_missing_artifact_file_is_a_resolution_error(tmp_path): + reg = FakeRegistry() + reg.add("qwen-base", 1, str(tmp_path / "gone.gguf"), {"kind": "full-model"}, serving=True) + with pytest.raises(serving_llm.ResolutionError, match="not found"): + serving_llm.resolve(store=_pointed_store("qwen-base"), client=reg) + + +def test_registry_outage_is_unavailable_not_an_error(): + class DownClient: + def get_model_version_by_alias(self, *a): + raise ConnectionError("registry down") + + with pytest.raises(serving_llm.ResolutionUnavailable): + serving_llm.resolve(store=_pointed_store("x"), client=DownClient()) + + +# -- T473: the adapter (base + LoRA) path ----------------------------------------------------------- + +def test_adapter_resolves_base_by_registered_name(tmp_path): + base, adapter = _gguf(tmp_path, "base.gguf"), _gguf(tmp_path, "adapter.gguf") + reg = FakeRegistry() + reg.add("qwen-base", 1, base, {"kind": "full-model"}, serving=True) + reg.add("ops-bot", 3, adapter, + {"kind": "lora-adapter", "base_model": "qwen-base"}, serving=True) + out = serving_llm.resolve(store=_pointed_store("ops-bot"), client=reg) + assert out["base_gguf"] == base and out["adapter_gguf"] == adapter + assert out["base"] == {"name": "qwen-base", "version": "1", "source": base} + + +def test_adapter_resolves_raw_hf_base_via_base_id_tag(tmp_path): + # The trainer stamps the RAW HF id (Qwen/…) — matched via the registered base's base_id tag + # (scripts/register_base_gguf.py), not a registered-model name. + base, adapter = _gguf(tmp_path, "base.gguf"), _gguf(tmp_path, "adapter.gguf") + reg = FakeRegistry() + reg.add("qwen2.5-0.5b-instruct", 1, base, + {"kind": "full-model", "base_id": "Qwen/Qwen2.5-0.5B-Instruct"}) + reg.add("ops-bot", 2, adapter, + {"kind": "lora-adapter", "base_model": "Qwen/Qwen2.5-0.5B-Instruct"}, serving=True) + out = serving_llm.resolve(store=_pointed_store("ops-bot"), client=reg) + assert out["base_gguf"] == base and out["adapter_gguf"] == adapter + + +def test_legacy_adapter_without_kind_is_inferred(tmp_path): + # FR-267: a pre-022 fine-tune (base_model + format=gguf, no kind) still serves as an adapter. + base, adapter = _gguf(tmp_path, "base.gguf"), _gguf(tmp_path, "adapter.gguf") + reg = FakeRegistry() + reg.add("qwen-base", 1, base, {"kind": "full-model"}) + reg.add("ops-bot", 1, adapter, {"format": "gguf", "base_model": "qwen-base"}, serving=True) + out = serving_llm.resolve(store=_pointed_store("ops-bot"), client=reg) + assert out["kind"] == "lora-adapter" and out["adapter_gguf"] == adapter + + +def test_missing_base_is_a_resolution_error(tmp_path): + reg = FakeRegistry() + reg.add("ops-bot", 1, _gguf(tmp_path, "a.gguf"), + {"kind": "lora-adapter", "base_model": "nowhere-base"}, serving=True) + with pytest.raises(serving_llm.ResolutionError, match="register the local base"): + serving_llm.resolve(store=_pointed_store("ops-bot"), client=reg) + + +def test_adapter_chain_is_refused_not_walked(tmp_path): + # contracts/serving-resolution.md: base_model MUST resolve DIRECTLY to a full-model version — + # an adapter pointing at another adapter is an error, never a multi-hop chain (PR #64 R2). + reg = FakeRegistry() + reg.add("mid-adapter", 1, _gguf(tmp_path, "mid.gguf"), + {"kind": "lora-adapter", "base_model": "qwen-base"}) + reg.add("ops-bot", 1, _gguf(tmp_path, "top.gguf"), + {"kind": "lora-adapter", "base_model": "mid-adapter"}, serving=True) + with pytest.raises(serving_llm.ResolutionError, match="no adapter chains"): + serving_llm.resolve(store=_pointed_store("ops-bot"), client=reg) + + +def test_adapter_without_base_lineage_is_refused(tmp_path): + reg = FakeRegistry() + reg.add("ops-bot", 1, _gguf(tmp_path, "a.gguf"), {"kind": "lora-adapter"}, serving=True) + with pytest.raises(serving_llm.ResolutionError, match="no base_model lineage"): + serving_llm.resolve(store=_pointed_store("ops-bot"), client=reg) + + +def test_base_prefers_serving_alias_then_newest_full_model(tmp_path): + g1, g2, adapter = _gguf(tmp_path, "v1.gguf"), _gguf(tmp_path, "v2.gguf"), \ + _gguf(tmp_path, "ad.gguf") + reg = FakeRegistry() + reg.add("qwen-base", 1, g1, {"kind": "full-model"}, serving=True) # @serving on v1 + reg.add("qwen-base", 2, g2, {"kind": "full-model"}) # newer, not promoted + reg.add("ops-bot", 1, adapter, + {"kind": "lora-adapter", "base_model": "qwen-base"}, serving=True) + out = serving_llm.resolve(store=_pointed_store("ops-bot"), client=reg) + assert out["base"]["version"] == "1" # the promoted base wins over the newer version + + +# -- llmresolve unit seams --------------------------------------------------------------------------- + +def test_task_from_kind_infers_only_llm_shapes(): + assert llmresolve.task_from_kind({"kind": "lora-adapter"}) == "text-generation" + assert llmresolve.task_from_kind({"format": "gguf"}) == "text-generation" + assert llmresolve.task_from_kind({"kind": "vision-classifier"}) is None + assert llmresolve.task_from_kind({}) is None + + +def test_effective_kind_legacy_inference(): + assert llmresolve.effective_kind({"kind": "full-model"}) == "full-model" + assert llmresolve.effective_kind({"kind": "lora-adapter"}) == "lora-adapter" + assert llmresolve.effective_kind( + {"base_model": "x", "format": "gguf"}) == "lora-adapter" # the pre-022 fine-tune shape + assert llmresolve.effective_kind({}) == "full-model" + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-q"])) diff --git a/training/flows/finetune.py b/training/flows/finetune.py index eaadcc9..61684e8 100644 --- a/training/flows/finetune.py +++ b/training/flows/finetune.py @@ -230,9 +230,14 @@ def register_version(output_name: str, gguf_path: str, run_id: str, # LLM flow) records genealogy identically (FR-095). `lineage=base` — LLM chaining from a *registered* # version isn't supported (the stored artifact is the serving GGUF, not a trainable PEFT adapter), so # there is no `parent_version` here; the resumable modalities are vision + embeddings. + # 022 T476 (FR-266): stamp task + serving_engine at registration — mirroring what vision/embed + # already tag — so a new fine-tune is a first-class, selectable text-generation serving target + # (never a task:null "no renderer" placeholder). The base_model lineage is what the serving + # resolver maps to a registered base GGUF (contracts/serving-resolution.md). mv = c.create_model_version( name=output_name, source=source, run_id=run_id, tags={"kind": "lora-adapter", "format": "gguf", + "task": "text-generation", "serving_engine": "llama.cpp", **lineage_tags(base_model, dataset_name, dataset_version)}, ) _log(f"registered {output_name} v{mv.version} <- run {run_id} on {dataset_name}@{dataset_version}") diff --git a/ui/components/models/PromoteGate.tsx b/ui/components/models/PromoteGate.tsx index 6839a4e..81f1fdb 100644 --- a/ui/components/models/PromoteGate.tsx +++ b/ui/components/models/PromoteGate.tsx @@ -7,11 +7,19 @@ // typed reason. Accepts the ?override=@ deep-link from the retraining inbox: the // candidate row is highlighted and framed for override review (the gate still runs — nothing // auto-fires). +// +// 022 T482 (FR-255/258/269): for a TEXT-GENERATION version the promote action IS the served-LLM +// switch (Clarifications 2026-07-05 — no separate "set serving" control). When the switch would +// displace a RESIDENT serving model, it is gated behind the shared ConfirmDialog NAMING the model +// to be displaced, and the promote is sent with `preempt: true` (the agent refuses displacement +// without it). The card shows the resident-vs-promoted delta, and the promote response's +// `serving_llm.reload` outcome (live / deferred with the agent's reason) is surfaced. -import { useState } from 'react'; +import { useEffect, useState } from 'react'; import { Badge } from '@/components/Badge'; import { ConfirmDialog } from '@/components/ConfirmDialog'; -import { gwPost } from '@/lib/gw'; +import type { ServingState } from '@/components/serving/types'; +import { gwGet, gwPost } from '@/lib/gw'; import { LineageLinks } from './LineageLinks'; export type Version = { @@ -34,7 +42,27 @@ export type Verdict = { incumbent: MetricBrief; delta: number | null; }; -type PromoteResult = { promoted: boolean; serving_version: string | null; verdict: Verdict }; +// 022: the go-live half of an LLM promote (pointer + agent reload) riding the promote response. +type ServingLLMResult = { + active: string | null; + kind?: string; + base?: { name: string; version: string; source: string } | null; + reload?: { status: string; reason?: string; evicted?: string | null }; + error?: string; +}; +type PromoteResult = { + promoted: boolean; + serving_version: string | null; + verdict: Verdict; + serving_llm?: ServingLLMResult; +}; + +// A text-generation version (task tag, or the artifact-kind inference for legacy versions). +function isLLMVersion(tags: Record): boolean { + return ( + tags?.task === 'text-generation' || tags?.kind === 'lora-adapter' || tags?.format === 'gguf' + ); +} type CompareLeg = { version?: string; metric?: string; value?: number } | null; type CompareRes = { champion?: CompareLeg; @@ -75,6 +103,19 @@ export function PromoteGate({ const [verdict, setVerdict] = useState(null); const [preview, setPreview] = useState<{ version: string; res: CompareRes } | null>(null); const [askOverride, setAskOverride] = useState(null); // version pending override + // 022: the served-LLM switch — the version pending the displacement confirm + who it displaces. + const [askSwitch, setAskSwitch] = useState<{ version: string; displaced: string } | null>(null); + const [servingLLM, setServingLLM] = useState(null); + const [liveState, setLiveState] = useState(null); + + const hasLLM = versions.some((v) => isLLMVersion(v.tags)); + useEffect(() => { + // resident-vs-promoted delta (FR-269) — only fetched for models with LLM versions + if (!hasLLM) return; + gwGet('serving/state') + .then(setLiveState) + .catch(() => setLiveState(null)); + }, [hasLLM, verdict]); const doPreview = async (version: string) => { setBusy(version); @@ -93,16 +134,19 @@ export function PromoteGate({ } }; - const promote = async (version: string, override = false) => { + const promote = async (version: string, override = false, preempt = false) => { setBusy(version); setErr(''); setPreview(null); + setServingLLM(null); try { const res = await gwPost(`models/${encodeURIComponent(name)}/promote`, { version, override, + preempt, }); setVerdict(res.verdict); + if (res.serving_llm) setServingLLM(res.serving_llm); onChanged(); } catch (e) { setErr(String(e)); @@ -111,10 +155,73 @@ export function PromoteGate({ } }; + // 022 (FR-258): promoting an LLM version that would displace a RESIDENT serving model goes + // through the ConfirmDialog naming the displaced model; an idle GPU promotes straight through. + const promoteLLMAware = async (v: Version) => { + if (!isLLMVersion(v.tags)) return promote(v.version); + let state: ServingState | null = null; + try { + state = await gwGet('serving/state'); // fresh — not the polled copy + } catch { + state = null; + } + setLiveState(state); + if (state?.resident) { + const displaced = + state.holder === 'llm' + ? `${state.serving_model}${state.serving_version ? ` v${state.serving_version}` : ''}` + : `the resident ${state.holder} engine`; + setAskSwitch({ version: v.version, displaced }); + return; + } + return promote(v.version); // idle GPU — the switch displaces nothing + }; + return ( <> {err &&

[x] {err}

} + {/* 022 FR-269: the resident-vs-promoted delta for an LLM model — what actually runs vs + what @serving names. Agent-reported ("unknown" when the agent is unreachable). */} + {hasLLM && liveState && ( +

+ [≡] live LLM:{' '} + + {liveState.serving_model} + {liveState.serving_version ? ` v${liveState.serving_version}` : ''} + {' '} + ({liveState.resident ? 'resident' : 'idle'}) + {championVersion && ( + <> + {' '} + · promoted here: v{championVersion} + + )} + {liveState.adapter && <> · adapter {liveState.adapter} on {liveState.base}} +

+ )} + + {/* 022: the go-live outcome of the last LLM promote — live at once, or deferred with the + agent's reason (job holder, missing confirm), or a pointer error. Never silent. */} + {servingLLM && ( +

+ {servingLLM.error ? ( + [x] switch: {servingLLM.error} + ) : servingLLM.reload && + ['loaded', 'reloaded', 'swapped', 'noop'].includes(servingLLM.reload.status) ? ( + + [⇄] switch: {servingLLM.reload.status === 'noop' ? 'already live' : 'live'} + {servingLLM.reload.evicted ? ` (displaced ${servingLLM.reload.evicted})` : ''} + + ) : ( + + [~] switch {servingLLM.reload?.status ?? 'pending'} + {servingLLM.reload?.reason ? `: ${servingLLM.reload.reason}` : ''} + + )} +

+ )} + {preview && (

@@ -175,8 +282,13 @@ export function PromoteGate({ [⇄] preview