From 86a6f3d5bab5406e93dbe6f35c1d7ff774f5a087 Mon Sep 17 00:00:00 2001 From: webdevtodayjason Date: Wed, 23 Sep 2026 20:31:55 -0500 Subject: [PATCH] Apply a decision adapter's temperatures and warm its grammar on bind /v1/decide and /v1/systemone now read temperatures.json from the served model's directory in this node's store (resolved through registry.snapshot_dir_for) and divide each question's label logprobs by its kind's temperature before the softmax, in run_questions so both routes get it. Both responses carry a top-level calibration block {applied, temperatures}; "calibration": "raw" opts out. A model whose directory carries prompt_contract.json or temperatures.json is sent one minimal constrained question per kind (choice, noul, score) through ask_one as soon as its engine binds, from _wait_for_bind, in the background. Each compile time is logged and /api/status gains an instances list with warm per instance. The engine-call timeout goes from 180 s to 300 s, and a connected call that times out now reads as the grammar compiling, retry, rather than as an unreachable node. Closes #276 Closes #277 Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 4 +- README.md | 31 +++ ainode/api/decide.py | 322 +++++++++++++++++++++++++++++-- ainode/api/server.py | 15 ++ ainode/api/systemone.py | 59 +++--- ainode/models/api_routes.py | 19 ++ tests/test_decide_calibration.py | 262 +++++++++++++++++++++++++ tests/test_decide_warmup.py | 191 ++++++++++++++++++ tests/test_systemone.py | 4 +- 9 files changed, 861 insertions(+), 46 deletions(-) create mode 100644 tests/test_decide_calibration.py create mode 100644 tests/test_decide_warmup.py diff --git a/AGENTS.md b/AGENTS.md index b7b61b78..34fb7dcf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -28,8 +28,8 @@ State / architecture / decisions / "why": Obsidian Vault, `AINode` (cluster ops: - **The legacy 110-item decision path (`--backend ainode|chat|jev`) scores typed decisions against labels, and its confidence numbers are the product.** Accuracy is the weakest number in the block: a wrong answer at 0.95 is the failure mode, so every block carries the Brier score on the labeled option, an expected calibration error with the five-bin reliability table behind it, and the count of wrong answers surviving a 0.8 and a 0.9 gate. **A probability nobody reported is absent, never assumed** (such a row is in the accuracy, out of the calibration, and counted in `no_confidence`), **an item that failed is one row with an `error`** and never a wrong answer, and **cost is a posted vendor rate over reported tokens or `0`** for a local backend, never an estimate. `bench/decide/items.json` is repo data versioned next to its results: loading is strict, a malformed item is a load error rather than a skipped item, and a set's items must share one kind, question and option set because a set is one measurement. Backends are split `request()` / `parse()` as pure functions so `tests/test_bench_decide.py` pins every request shape and response shape with canned payloads and no network. **The TypeSafe key is never printed, never written into a record and never put in a note**: a run reports only which of `--api-key`, `$TYPESAFE_API_KEY` or `~/.jev_api_key` it came from. Sets, metrics and flags: `bench/decide/README.md`. - **The speech bench (`ainode/bench/speech/`, `scripts/ainode-bench.py speech`) scores against committed audio, and both halves of that are load-bearing.** A word error rate is only comparable over the same bytes, so the ten clips live in `bench/speech/clips/` as repo data (1.4 MB) rather than being synthesised per run, and **`clips.CLIPS_VERSION` is bumped on any edit to a text, a voice or a WAV**; `--generate-clips` rebuilds the set with macOS `say` plus `afconvert` and is a maintenance step a run never takes. The reference is the exact string handed to `say`, fixed before the run, and **nothing adjusts a reference after a transcript is seen**: a reference edited to match what a model said makes the rate a statement about the editor. The normaliser is part of the measurement, so it is versioned (`metrics.NORMALIZER_VERSION`) and the record carries BOTH rates, `wer` with number words folded to digits and `wer_orthographic` with case and punctuation only, because a transcript that heard every word and wrote "9" for "nine" is not a hearing error and one number alone hides which kind it was. Nothing is folded that changes a word: no stopword list, no stemming, no synonym map, no per-clip exception. `wer` is pooled over words, never a mean of per-clip rates. **A clip that failed is one row with an `error` and nulls for every number**, counted out of every rate, percentile and factor, never folded in as a 100 percent error rate: a transport failure inside a figure a reader takes as the model's is the one mistake this section can make. It is the one bench whose request body is not JSON (`client.py` assembles the multipart itself), so a run through a node's `:3000/v1` exercises the fleet's own audio path; the block is `bench/SCHEMA.md`. - **One proxy handler serves every forwarded inference path**: `proxy_to_vllm` is registered for `POST /v1/chat/completions`, `/v1/completions`, `/v1/messages`, `/v1/messages/count_tokens`, `/v1/responses`, `/v1/rerank`, `/v1/score`, `/v1/audio/transcriptions`, `/v1/audio/translations`, `/tokenize` and `/detokenize` (`GET /v1/models` is the federated union and does not forward; `POST /v1/embeddings` keeps its own handler because it validates the body first, and `/v1/decide` plus `/v1/systemone` compose their own completions). Add a path by registering it on that handler, never by writing a second proxy: routing on the body's `model`, transport failover, the multimodal ordering below, SSE passthrough and header passthrough are all protocol-agnostic and already there. Two of the paths are NOT under `/v1` because vLLM does not serve them there (`/tokenize`, `/detokenize`), so the table is the authority on the path and not a prefix rule. **The two audio paths are the ones whose body is NOT JSON**: OpenAI's speech-to-text API is a `multipart/form-data` upload with the model id as a form field, so the handler reads it with `api/multipart.py::form_fields` over the body it already buffered (never `request.multipart()`, which consumes the stream the proxy still has to forward) and forwards the bytes UNCHANGED under the caller's own `Content-Type`: a multipart body is only parseable against the boundary in its own header, so re-encoding the parts hands the engine a body the forwarded header no longer describes. A multipart body that names no `model` is a 400 naming the field, never a fallback to this node's own model, which would send someone's audio to a chat engine. They are also the first pair whose existence is MODEL-CONDITIONAL: vLLM attaches its speech-to-text router only when the served model reports the `transcription` task, so no chat or pooling engine's `/openapi.json` lists them and the check below cannot be run on one. For a path like that, the evidence is the router's own declaration in the engine image the recipe pins (`entrypoints/openai/speech_to_text/api_router.py`), and the `openapi.json` check still applies the first time such a model actually serves. Forward `request.path_qs`, not `request.path`: Claude Code posts to `/v1/messages?beta=true`. This is a route table and not a catch-all: an unregistered path stays a 404, which is why a new path is added only after `curl http://:/openapi.json` on a real engine says the engine answers it. -- **`/v1/decide`'s response shape is a contract, not an implementation detail** (`api/decide.py`). Its bench is written against the exact shape (`model`, `node`, `latency_ms`, `decisions[key] = {answer, confidence, distribution, latency_ms}`, `usage = {prompt_tokens, completion_tokens, calls}`), so a `200` always carries every question asked: a bad request is a `400` and an engine that cannot answer is a `503`, never a partial `decisions` block. Probabilities are keyed by the caller's OPTION strings, never by the letters used to constrain the engine. It is the one `/v1` path deliberately NOT on `proxy_to_vllm`, because it composes N grammar-constrained chat completions of its own from one request and has no caller body to forward; it still routes through the proxy's own `_routing_candidates` and the shared `app["client_session"]`, so never give it its own routing rule or HTTP stack. The constraint field is vLLM 0.27.1's `structured_outputs: {"choice": [...]}`: the legacy `guided_choice` is accepted by that image and then silently ignored, so sending it instead would produce free prose with no error. The engine-facing half of a decision request is `decide.py::run_questions`, shared with `/v1/systemone`: one path to the engines and one way the probabilities are read, so a route decides the status and the shape and nothing else. -- **`/v1/systemone` is TypeSafe's wire format and nothing else** (`api/systemone.py`). It exists so a client written for the hosted System One endpoint (Titanium's JDE `jevJudge({endpoint, model})`, browser-use's jev-ultrafast, the TypeSafe SDK, the playground) answers off a model on this fleet with its endpoint changed and nothing else, which makes the request and response shapes THEIRS and not ours to tidy: `{state, model, questions}` in, `{model, answers, usage: {input_tokens, output_tokens}, latency_ms}` out, a `choice` answering with the caller's own criteria key, a `noul` answering P(true), a `score` answering the expected level with a `legend` from position to level name, a `422` naming the field on a malformed request and a `503` when no node serves the model. JDE's `parseAnswers` discards the WHOLE answer set over one answer it cannot read, so every probability is finite and inside [0, 1] (`systemone.py::probability`) and a question the engine left unanswered is a `503`, never a 200 carrying an invented option. **`confidence` is NOT the picked option's probability**: the hosted service reports it chance corrected, `(n * p_max - 1) / (n - 1)`, which is INFERRED from every example in TypeSafe's published docs and SDK types (`systemone.py::normalized_confidence`) because no document states it, and a caller's bands are tuned against those numbers; the raw distribution goes out untouched beside it, so the formula is revisable against evidence and the numbers it came from are never lost. **A question may carry at most `TOP_LOGPROBS` criteria**, which is the engine request's ceiling and not the format's 255: only the top 20 labels come back with a probability, so a wider option set is a `422` naming the cap rather than a distribution missing its tail, and raising it means a second pass over the remaining labels, which is a measurement and not a constant. Translate in and out around `run_questions`: a gap here is never answered with a second engine path, a second routing rule or a second reading of the logprobs. **Calibration is the model's**, so nothing on this path rescales, tempers or corrects what the engine reported, and the way to find out what a local model's confidence is worth stays `scripts/ainode-bench.py decide`. `tests/test_systemone.py` carries JDE's own reader rule for rule and replays a real case from its blind set, read where it lives (`JDE_COMPLETION_CASES`, default `/Users/sem/code/jde/cases/`) and never copied in: it is someone else's measurement data, and the test skips where it is absent. +- **`/v1/decide`'s response shape is a contract, not an implementation detail** (`api/decide.py`). Its bench is written against the exact shape (`model`, `node`, `latency_ms`, `decisions[key] = {answer, confidence, distribution, latency_ms}`, `usage = {prompt_tokens, completion_tokens, calls}`), so a `200` always carries every question asked: a bad request is a `400` and an engine that cannot answer is a `503`, never a partial `decisions` block. Probabilities are keyed by the caller's OPTION strings, never by the letters used to constrain the engine. It is the one `/v1` path deliberately NOT on `proxy_to_vllm`, because it composes N grammar-constrained chat completions of its own from one request and has no caller body to forward; it still routes through the proxy's own `_routing_candidates` and the shared `app["client_session"]`, so never give it its own routing rule or HTTP stack. The constraint field is vLLM 0.27.1's `structured_outputs: {"choice": [...]}`: the legacy `guided_choice` is accepted by that image and then silently ignored, so sending it instead would produce free prose with no error. The engine-facing half of a decision request is `decide.py::run_questions`, shared with `/v1/systemone`: one path to the engines and one way the probabilities are read, so a route decides the status and the shape and nothing else. **A decision adapter's own temperatures are applied there and nowhere else** (#276): when the served model's directory in THIS node's store (`decide.py::model_store_dir`, which goes through `models/registry.py::snapshot_dir_for`, never a hardcoded path) carries `temperatures.json`, each question's label logprobs are divided by its kind's temperature (`choice`, `noul`, `score`; decide's `boolean` is `noul`, its `score` is `score`, an `options` list is `choice`) before the softmax, `"calibration": "raw"` opts out, and both routes answer a top-level `calibration: {applied, temperatures}` block so a caller can refit. **A decision model warms on bind** (#277): `models/api_routes.py::_wait_for_bind` hands every bound engine to `decide.py::schedule_decision_warmup`, which sends one constrained question per kind through `ask_one` in the background when the directory carries `prompt_contract.json` or `temperatures.json`, and `/api/status` reports `warm` per instance (null for a model with nothing to warm). Readiness is not held for it. A connected engine call that times out is reported as the grammar compiling, never as an unreachable node. +- **`/v1/systemone` is TypeSafe's wire format and nothing else** (`api/systemone.py`). It exists so a client written for the hosted System One endpoint (Titanium's JDE `jevJudge({endpoint, model})`, browser-use's jev-ultrafast, the TypeSafe SDK, the playground) answers off a model on this fleet with its endpoint changed and nothing else, which makes the request and response shapes THEIRS and not ours to tidy: `{state, model, questions}` in, `{model, answers, usage: {input_tokens, output_tokens}, latency_ms}` out (plus AINode's `calibration` block, #276), a `choice` answering with the caller's own criteria key, a `noul` answering P(true), a `score` answering the expected level with a `legend` from position to level name, a `422` naming the field on a malformed request and a `503` when no node serves the model. JDE's `parseAnswers` discards the WHOLE answer set over one answer it cannot read, so every probability is finite and inside [0, 1] (`systemone.py::probability`) and a question the engine left unanswered is a `503`, never a 200 carrying an invented option. **`confidence` is NOT the picked option's probability**: the hosted service reports it chance corrected, `(n * p_max - 1) / (n - 1)`, which is INFERRED from every example in TypeSafe's published docs and SDK types (`systemone.py::normalized_confidence`) because no document states it, and a caller's bands are tuned against those numbers; the raw distribution goes out untouched beside it, so the formula is revisable against evidence and the numbers it came from are never lost. **A question may carry at most `TOP_LOGPROBS` criteria**, which is the engine request's ceiling and not the format's 255: only the top 20 labels come back with a probability, so a wider option set is a `422` naming the cap rather than a distribution missing its tail, and raising it means a second pass over the remaining labels, which is a measurement and not a constant. Translate in and out around `run_questions`: a gap here is never answered with a second engine path, a second routing rule or a second reading of the logprobs. **Calibration is the model's**: the one correction on this path is the served adapter's own `temperatures.json`, applied in `run_questions` as the `/v1/decide` bullet says, and nothing here adds a correction of its own. The way to find out what a local model's confidence is worth stays `scripts/ainode-bench.py decide`. `tests/test_systemone.py` carries JDE's own reader rule for rule and replays a real case from its blind set, read where it lives (`JDE_COMPLETION_CASES`, default `/Users/sem/code/jde/cases/`) and never copied in: it is someone else's measurement data, and the test skips where it is absent. - **A multimodal chat request consults the capability cache before routing** (`api/server.py::proxy_to_vllm`). A body carrying an `image_url` / `input_audio` / `video_url` / `file` part, or the Anthropic Messages spelling (an `image` / `document` block, including one nested in a `tool_result`), is never routed on the model id alone: order candidates accepting-first (cached `vision: true`), then never-probed, and drop instances cached `vision: false`; vLLM's `may be provided in one prompt` 400 is a routing miss, so record `vision: false` and fail over, while every other 4xx goes back to the caller untouched. A request with no media keeps the plain order (local hop first, then peers). There is ONE capability cache: `app["chat_caps_cache"]`, filled by `/api/models/caps` in `api/chat_routes.py`, which probes remote instances directly on their engine port. Never add a second. - **Every node-to-node request AINode makes carries the FLEET KEY, and there is one helper that puts it there** (`ainode/auth/fleet.py`: `fleet_headers(app)` off a running app, `fleet_key_headers(secret)` off a config). The key is `HMAC-SHA256(cluster_secret, "ainode-fleet-key-v1")`, so every node holding the secret computes the same one, the join flow already distributes it, rotation follows the secret, and NOTHING new is written to disk. The middleware accepts it as the caller id `fleet` (`auth/middleware.py::identify_caller`), reading the secret LIVE per request, so a node whose secret differs refuses its would-be peers exactly as it drops their datagrams. Auth was enabled-able and unusable on a cluster before this: a node with `auth.enabled` answered 401 to its own fan-outs, so the fleet ran open. Add a peer call and you add the header: an ENGINE port is not a peer call in this sense (the inference proxy, the capability probes and the embeddings route talk to a vLLM container, which never sees this middleware), and `POST /api/cluster/join` is keyless by construction. `tests/test_fleet_auth.py` WALKS THE SOURCE for peer URLs and fails on one whose function does not name the helper, with an exempt list that has to state a reason. - **A fresh install requires a key, and an update never changes an installed node's access control** (`scripts/install.sh`). The installer mints one key on a node with no `config.json`, stores the SHA-256 the way `AuthConfig` does, prints the plaintext ONCE in a box, and the summary reads "API protected, one key"; `AINODE_AUTH=off` keeps the old open behaviour and prints what that choice means. Both gates matter: minting on an existing home would lock out every client the operator already pointed at the node, with a key they never saw. Because of this, `ainode update` verifies the running release on `/api/health` and not `/api/status` (health is the one keyless route, which is why it carries `version`): reading a keyed route there made every update on a protected node pull, pin, restart and then report that it had not applied. diff --git a/README.md b/README.md index 6b2eaaf3..87087bee 100644 --- a/README.md +++ b/README.md @@ -1278,6 +1278,37 @@ tokenizer does not give a two-letter label its own token, that label's mass is its first letter's token mass, which the single-letter label of the same letter also claims. Read such a pair as jointly calibrated. +**Decision adapters bring their own temperatures.** When the served model's +directory in the node's store carries a `temperatures.json` +(`{"temperatures": {"choice": T, "noul": T, "score": T}}`, which the +`frontier-infra/jebadiah-*` checkpoints ship), each question's label logprobs are +divided by the temperature for its kind before the softmax, so `distribution` +and `confidence` read the way the adapter was fitted. `type: "boolean"` counts as +`noul`, `type: "score"` as `score` and an `options` list as `choice`. Every +response says what was applied, and `POST /v1/systemone` carries the same block +(these temperatures are illustrative, not fitted): + +```json +"calibration": {"applied": true, "temperatures": {"choice": 1.5, "noul": 0.8, "score": 1.2}} +``` + +Send `"calibration": "raw"` to get the engine's own spread instead; the block +then reads `"applied": false` and still lists the temperatures you opted out of, +so you can refit against the raw numbers. The file is read on the node that +answers the route, so a model this node routes to a peer without holding a copy +itself answers raw. + +**A decision model warms up when it loads.** The first constrained request per +question shape makes the engine compile the answer grammar, 60 to 90 s on a +GB10. So when an engine binds on a model whose directory carries +`prompt_contract.json` or `temperatures.json`, AINode sends it one small +question of each kind in the background and logs each compile time. Each row of +`/api/status`'s `instances` carries `warm` (`true` once warm, `false` while +warming or after a failed warm-up, `null` for a model with nothing to warm) and +`warm_compile_seconds`. A request that still meets a cold compile and runs past +the 300 s engine-call limit gets a `503` saying the engine is compiling the +answer grammar and to retry. + ### Metrics: `/metrics` (Prometheus) and `/api/metrics` (JSON) AINode exposes its own metrics on port 3000, the same port as its API. The engine's diff --git a/ainode/api/decide.py b/ainode/api/decide.py index aeea8c96..e0341ab8 100644 --- a/ainode/api/decide.py +++ b/ainode/api/decide.py @@ -29,6 +29,15 @@ engine's own refusal to produce anything but a label is what makes that restriction sound. +Calibration, when the model ships its own: a decision adapter's store directory +can carry ``temperatures.json`` beside the weights, one fitted temperature per +question kind (``choice``, ``noul``, ``score``). When the served model's +directory on this node has one, the label logprobs are divided by that kind's +temperature before the softmax, so the distribution and the confidence read the +way the adapter was fitted to be read. A request may opt out with +``"calibration": "raw"``, and every response says what was applied so a caller +can refit against the raw numbers. + vLLM field note: on the pinned engine image (``vllm/vllm-openai:v0.27.1``) the constraint is ``structured_outputs: {"choice": [...]}``. The older ``guided_choice`` extra field is still accepted and then SILENTLY IGNORED on @@ -42,7 +51,9 @@ import asyncio import json import math +import logging import time +from pathlib import Path from typing import Any, NamedTuple, Optional import aiohttp @@ -50,6 +61,8 @@ from ainode.api.chat_routes import instance_caps_index +logger = logging.getLogger(__name__) + SYSTEM_PROMPT = ("You are a decision function. Answer with the single letter " "of the best option and nothing else.") @@ -70,9 +83,15 @@ # Room for the longest label plus the end-of-turn token the template emits. LABEL_TOKEN_HEADROOM = 1 -# A question's engine call. Long enough for a cold-ish engine to answer two -# tokens, short enough that a wedged node does not hold the whole request open. -CALL_TIMEOUT_S = 180.0 +# A question's engine call. Long enough for a cold engine to compile the answer +# grammar for a new question shape, which measured 60 to 90 s per shape on a GB10 +# and can queue behind the other questions of the same request (#277), short +# enough that a wedged node does not hold the whole request open for good. +CALL_TIMEOUT_S = 300.0 + +# The warm-up's engine call, per question kind. A first compile on a busy node is +# allowed to take a long time: nobody is waiting on it. +WARM_TIMEOUT_S = 600.0 # A dead or ghost node must fail the connect fast so failover moves on. CONNECT_TIMEOUT_S = 5.0 @@ -82,6 +101,24 @@ DEFAULT_SCORE_MIN = 1 DEFAULT_SCORE_MAX = 5 +# The question kinds a decision adapter is fitted per. ``/v1/systemone`` speaks +# them natively; ``/v1/decide`` maps ``type: "boolean"`` to ``noul``, +# ``type: "score"`` to ``score`` and an explicit ``options`` list to ``choice``. +CHOICE = "choice" +NOUL = "noul" +SCORE = "score" +QUESTION_KINDS = (CHOICE, NOUL, SCORE) + +# What marks a decision model's store directory: the adapter's prompt contract +# and its fitted temperatures, either one. Only the second is read here. +PROMPT_CONTRACT_FILE = "prompt_contract.json" +TEMPERATURES_FILE = "temperatures.json" +DECISION_MODEL_FILES = (PROMPT_CONTRACT_FILE, TEMPERATURES_FILE) + +# The one value a request's ``calibration`` field takes: the distribution exactly +# as the engine reported it, with the adapter's temperatures left off. +CALIBRATION_RAW = "raw" + class DecideError(Exception): """A bad request shape. Carries the message the caller gets in the 4xx. @@ -167,6 +204,7 @@ def normalize_questions(raw: Any) -> dict[str, dict]: if not isinstance(text, str) or not text.strip(): raise DecideError(f"question '{key}' needs a non-empty 'question' string") qtype = spec.get("type") + kind = NOUL if qtype == "boolean" else SCORE if qtype == "score" else CHOICE if "options" in spec: options = spec["options"] elif qtype == "boolean": @@ -198,7 +236,7 @@ def normalize_questions(raw: Any) -> dict[str, dict]: raise DecideError( f"question '{key}': duplicate options {dupes}. Every option must " f"be distinct so an answer is unambiguous") - out[key] = {"question": text.strip(), "options": list(options)} + out[key] = {"question": text.strip(), "options": list(options), "kind": kind} return out @@ -275,10 +313,14 @@ def first_token_top_logprobs(payload: dict) -> list[dict]: return [t for t in tops if isinstance(t, dict) and isinstance(t.get("token"), str)] -def distribution_from_logprobs(labels: list[str], - top_logprobs: list[dict]) -> Optional[dict]: +def distribution_from_logprobs(labels: list[str], top_logprobs: list[dict], + temperature: float = 1.0) -> Optional[dict]: """Softmax over the first token's logprobs, restricted to the label tokens. + ``temperature`` divides the logprobs before the softmax, which is the same as + dividing the logits: the log-normalizer is one constant across the labels and + drops out when the result is renormalized. 1.0 is the engine's own spread. + A label is scored by the LONGEST token in ``top_logprobs`` that is a prefix of it, which is an exact match whenever the tokenizer gives the whole label one token (the common single-letter case, and ``AB`` on the Ornith @@ -323,7 +365,8 @@ def distribution_from_logprobs(labels: list[str], return None top = max(scored.values()) - weights = {label: math.exp(lp - top) for label, lp in scored.items()} + weights = {label: math.exp((lp - top) / temperature) + for label, lp in scored.items()} total = sum(weights.values()) if total <= 0: return None @@ -362,17 +405,19 @@ def constrained_label(payload: dict, labels: list[str]) -> Optional[str]: def decision_from_payload(payload: dict, options: list[str], - latency_ms: float) -> dict: + latency_ms: float, temperature: float = 1.0) -> dict: """One ``decisions`` entry from one engine response. Pure. Probabilities are reported against the OPTION strings, not the labels: the labels are an implementation detail of constraining the engine, and a caller - that had to map them back would be doing our job. + that had to map them back would be doing our job. ``temperature`` is the + adapter's fitted temperature for this question's kind, 1.0 for none. """ labels = option_labels(len(options)) by_label = dict(zip(labels, options)) chosen = constrained_label(payload, labels) - dist = distribution_from_logprobs(labels, first_token_top_logprobs(payload)) + dist = distribution_from_logprobs(labels, first_token_top_logprobs(payload), + temperature) answer_label = pick_answer(labels, dist, chosen) entry: dict = { "answer": by_label.get(answer_label), @@ -391,6 +436,91 @@ def decision_from_payload(payload: dict, options: list[str], return entry +# ---------------------------------------------------------------- calibration + + +def calibration_mode(value: Any) -> Optional[str]: + """A request's ``calibration`` field: absent for the adapter's, or ``"raw"``.""" + if value is None or value == CALIBRATION_RAW: + return value + raise DecideError(f"'calibration' must be \"{CALIBRATION_RAW}\" when given " + f"(got {value!r})") + + +def model_store_dir(models_dir, model: str) -> Optional[Path]: + """The directory holding ``model``'s files in this node's store, or None. + + The store's own resolver (``models/registry.py::snapshot_dir_for``) answers + for a repo id in every layout AINode writes; a model served straight from an + absolute path is its own directory. + """ + from ainode.models.registry import snapshot_dir_for + if not model: + return None + try: + direct = Path(model) + if direct.is_absolute() and direct.is_dir(): + return direct + except OSError: + pass + return snapshot_dir_for(model, Path(models_dir) if models_dir else None) + + +def is_decision_model_dir(directory: Optional[Path]) -> bool: + """True when the directory carries a decision adapter's prompt contract or + temperatures, which is how a decision model is recognised.""" + if directory is None: + return False + try: + return any((Path(directory) / name).is_file() for name in DECISION_MODEL_FILES) + except OSError: + return False + + +def read_temperatures(directory: Optional[Path]) -> Optional[dict]: + """``{kind: T}`` from the directory's ``temperatures.json``, or None. + + Only the three kinds are read, and only a finite positive number is a + temperature: anything else leaves that kind at the engine's own spread rather + than failing a request over an adapter's file. + """ + if directory is None: + return None + try: + raw = json.loads((Path(directory) / TEMPERATURES_FILE).read_text()) + except (OSError, ValueError): + return None + temps = raw.get("temperatures") if isinstance(raw, dict) else None + if not isinstance(temps, dict): + return None + out: dict[str, float] = {} + for kind in QUESTION_KINDS: + value = temps.get(kind) + if isinstance(value, bool): + continue + try: + t = float(value) + except (TypeError, ValueError): + continue + if math.isfinite(t) and t > 0: + out[kind] = t + return out or None + + +def calibration_for(models_dir, model: str, mode: Optional[str]) -> dict: + """The ``calibration`` block a response carries: ``{applied, temperatures}``. + + ``temperatures`` is what the model's directory on this node carries, whether + or not it was applied, so a caller that opted out still sees what it opted + out of; it is null when there is none. The directory is looked up on the node + answering the route: a model this node routes to a peer and does not hold a + copy of answers raw, and says so with ``applied: false``. + """ + temps = read_temperatures(model_store_dir(models_dir, model)) + return {"applied": bool(temps) and mode != CALIBRATION_RAW, + "temperatures": temps} + + # -------------------------------------------------------------------- routing @@ -429,18 +559,22 @@ def candidates_for(request: web.Request, model: str) -> list: return candidates -async def ask_one(session: aiohttp.ClientSession, candidates: list, body: dict - ) -> tuple: +async def ask_one(session: aiohttp.ClientSession, candidates: list, body: dict, + timeout_s: float = CALL_TIMEOUT_S) -> tuple: """One question, with the proxy's failover. Returns (payload, cand, latency_ms). A transport failure or a 5xx moves to the next candidate: a ghost node that still advertises the model is indistinguishable from a live one in cluster state. Any other non-200 is the engine's own answer and stops the loop. On failure the third element is the error string instead of a latency. + + A call that connected and then ran out of time is reported as what it almost + always is, an engine compiling the answer grammar for a question shape it has + not seen (#277), rather than as an unreachable node. """ started = time.monotonic() last_err = "no candidate node" - timeout = aiohttp.ClientTimeout(total=CALL_TIMEOUT_S, + timeout = aiohttp.ClientTimeout(total=timeout_s, sock_connect=CONNECT_TIMEOUT_S) for host, port in candidates: url = f"http://{host}:{port}/v1/chat/completions" @@ -459,7 +593,15 @@ async def ask_one(session: aiohttp.ClientSession, candidates: list, body: dict last_err = f"{host}:{port} answered {resp.status}: {str(text)[:200]}" if 400 <= resp.status < 500: break - except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + except aiohttp.ServerTimeoutError as exc: + # The connect itself timed out (no read timeout is set): a dead node. + last_err = f"{host}:{port} unreachable: {exc}" + continue + except asyncio.TimeoutError: + last_err = (f"{host}:{port} gave no answer in {timeout_s:.0f}s: the engine " + "is compiling the answer grammar, retry") + continue + except aiohttp.ClientError as exc: last_err = f"{host}:{port} unreachable: {exc}" continue return None, None, last_err @@ -494,18 +636,21 @@ class DecideRun(NamedTuple): the caller keyed the question and in the caller's order; ``payloads`` are the raw engine responses, for the usage block; ``landed`` is the candidate the first answer came from, for naming the node; ``failures`` is one line per - question that got no answer. + question that got no answer; ``calibration`` is the ``{applied, + temperatures}`` block both routes report. """ decisions: dict payloads: list landed: Optional[tuple] failures: list + calibration: Optional[dict] = None async def run_questions(request: web.Request, model: str, questions: dict[str, dict], state: str, instructions: Optional[str], - candidates: list) -> DecideRun: + candidates: list, calibration: Optional[str] = None + ) -> DecideRun: """Ask every question at once against `candidates`, and read the answers. The whole engine-facing half of a decision request, shared by ``/v1/decide`` @@ -517,8 +662,17 @@ async def run_questions(request: web.Request, model: str, questions: dict[str, d All questions are in flight together. They share a byte-identical prompt prefix, so once one of them has prefilled it the rest read the shared state out of the engine's prefix cache instead of paying for it again. + + ``calibration`` is the request's mode (``calibration_mode``). Unless it is + ``"raw"``, each question's logprobs are divided by the temperature the + model's directory carries for that question's ``kind`` before the softmax. """ session: aiohttp.ClientSession = request.app["client_session"] + config = request.app.get("config") + block = await asyncio.get_running_loop().run_in_executor( + None, calibration_for, getattr(config, "models_dir", None), model, + calibration) + temps = block["temperatures"] if block["applied"] else {} keys = list(questions) bodies = [] for key in keys: @@ -541,9 +695,135 @@ async def run_questions(request: web.Request, model: str, questions: dict[str, d continue payloads.append(payload) landed = landed or cand - decisions[key] = decision_from_payload(payload, questions[key]["options"], - float(extra)) - return DecideRun(decisions, payloads, landed, failures) + decisions[key] = decision_from_payload( + payload, questions[key]["options"], float(extra), + temps.get(questions[key].get("kind"), 1.0)) + return DecideRun(decisions, payloads, landed, failures, block) + + +# -------------------------------------------------------------------- warm-up +# +# The first grammar-constrained request per question shape makes vLLM compile +# the answer grammar, 60 to 90 s on a GB10, and the route's caller is the one who +# waits for it (#277). So a decision model is sent one minimal question of each +# kind as soon as its engine binds, through the same body builder and the same +# ``ask_one`` the routes use, and the instance reports ``warm`` on /api/status. +# Readiness is not held for it: the engine serves while it warms. + +# Warm-ups in flight. The loop holds only a weak reference to a task. +_WARM_TASKS: set = set() + + +def warm_questions() -> dict[str, dict]: + """One minimal question per kind: a two-option choice, a noul, a two-level score. + + The noul's options are ``true`` / ``false`` in the order ``/v1/systemone`` + always sends them. + """ + return { + CHOICE: {"question": "Which option fits the state?", + "options": ["first", "second"]}, + NOUL: {"question": "Is the state empty?", "options": ["true", "false"]}, + SCORE: {"question": "How complete is the state?", "options": ["1", "2"]}, + } + + +def served_model_id(backend) -> str: + """The id an engine answers to: its first ``--served-model-name``, else its model.""" + cfg = getattr(backend, "config", None) + names = getattr(cfg, "served_model_name", None) or [] + if isinstance(names, str): + names = [names] + return str(names[0] if names else getattr(cfg, "model", "") or "") + + +async def warm_decision_engine(session: aiohttp.ClientSession, port: int, model: str, + status: dict, timeout_s: float = WARM_TIMEOUT_S + ) -> bool: + """Ask the engine on ``port`` one question per kind, in turn, and time each. + + One at a time so each compile is timed on its own and logged. ``status`` is + the instance's entry in ``app["decision_warm"]`` and is updated in place: + ``warm`` turns True only when every kind answered, ``compile_seconds`` holds + each kind's time and ``error`` the first failure. + """ + status.update(warm=False, warming=True, compile_seconds={}, error=None) + for kind, spec in warm_questions().items(): + labels = option_labels(len(spec["options"])) + messages = build_messages("", None, spec["question"], spec["options"]) + started = time.monotonic() + payload, _, extra = await ask_one(session, [("localhost", port)], + build_chat_body(model, messages, labels), + timeout_s) + seconds = round(time.monotonic() - started, 1) + if payload is None: + status.update(warming=False, error=f"{kind}: {extra}") + logger.warning("decision warm-up of %s on :%s failed at %s after %.1fs: %s", + model, port, kind, seconds, extra) + return False + status["compile_seconds"][kind] = seconds + logger.info("decision warm-up of %s on :%s: %s grammar ready in %.1fs", + model, port, kind, seconds) + status.update(warm=True, warming=False) + return True + + +async def _run_warmup(app, port: int, model: str, status: dict) -> bool: + try: + session = app.get("client_session") + if session is not None: + return await warm_decision_engine(session, port, model, status) + async with aiohttp.ClientSession() as own: + return await warm_decision_engine(own, port, model, status) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.exception("decision warm-up of %s on :%s raised", model, port) + status.update(warm=False, warming=False, error=str(exc)) + return False + + +def schedule_decision_warmup(app, port: int, backend) -> bool: + """Warm ``backend`` in the background if it serves a decision model. Called on bind. + + A decision model is one whose store directory carries ``prompt_contract.json`` + or ``temperatures.json``. Anything else is left alone and its entry cleared, + so a port that used to hold a decision model does not go on reporting it. + Returns True when a warm-up was started. + """ + table = app.get("decision_warm") if hasattr(app, "get") else None + if table is None or backend is None: + return False + cfg = getattr(backend, "config", None) + model = str(getattr(cfg, "model", "") or "") + models_dir = (getattr(cfg, "models_dir", None) + or getattr(app.get("config"), "models_dir", None)) + if not is_decision_model_dir(model_store_dir(models_dir, model)): + table.pop(port, None) + return False + status = {"model": model, "warm": False, "warming": True, + "compile_seconds": {}, "error": None} + table[port] = status + task = asyncio.get_running_loop().create_task( + _run_warmup(app, port, served_model_id(backend), status)) + _WARM_TASKS.add(task) + task.add_done_callback(_WARM_TASKS.discard) + return True + + +def instance_warm_fields(app, record) -> dict: + """``warm`` and ``warm_compile_seconds`` for one instance on /api/status. + + ``warm`` is True once every question kind has compiled, False while a + decision model is still warming or its warm-up failed, and null for a model + that is not a decision model (nothing to warm). + """ + table = app.get("decision_warm") or {} + entry = table.get(getattr(record, "api_port", None)) + if not entry or entry.get("model") != getattr(record, "model", None): + return {"warm": None, "warm_compile_seconds": None} + return {"warm": bool(entry.get("warm")), + "warm_compile_seconds": dict(entry.get("compile_seconds") or {})} # -------------------------------------------------------------------- handler @@ -579,6 +859,7 @@ async def handle_decide(request: web.Request) -> web.Response: instructions = body.get("instructions") if instructions is not None and not isinstance(instructions, str): raise DecideError("'instructions' must be a string when given") + calibration = calibration_mode(body.get("calibration")) except DecideError as exc: return _bad_request(str(exc)) @@ -593,7 +874,7 @@ async def handle_decide(request: web.Request) -> web.Response: return unavailable(f"no node is serving '{model}'") run = await run_questions(request, model, questions, state, instructions, - candidates) + candidates, calibration) collector = request.app.get("metrics_collector") total_ms = (time.monotonic() - started) * 1000 if run.failures: @@ -613,4 +894,5 @@ async def handle_decide(request: web.Request) -> web.Response: "latency_ms": round(total_ms, 1), "decisions": run.decisions, "usage": merge_usage(run.payloads), + "calibration": run.calibration, }) diff --git a/ainode/api/server.py b/ainode/api/server.py index 246f6416..d90f26a9 100644 --- a/ainode/api/server.py +++ b/ainode/api/server.py @@ -192,6 +192,9 @@ def create_app( app["instances"] = _seed app["start_time"] = time.time() app["client_session"] = None # lazy-init in startup + # Decision-model warm-up state by engine port (api/decide.py, #277). Created + # here because the app is frozen by the time an engine binds. + app["decision_warm"] = {} app["metrics_collector"] = collector # On a unified-memory node the collector has no usage figure of its own to # report, and host RAM is not VRAM (#175). Give it the one number this node @@ -1393,6 +1396,17 @@ async def handle_status(request: web.Request) -> web.Response: _own = _manager.by_port(getattr(config, "api_port", 0) or 0) engine_adopted = bool(_own is not None and getattr(_own.record, "adopted", False)) + # Every instance this process manages, with whether a decision model's answer + # grammar has been compiled yet (#277). ``warm`` is null for a model with + # nothing to warm. + from ainode.api.decide import instance_warm_fields + instances = [{"instance_id": inst.record.instance_id, + "model": inst.record.model, + "api_port": inst.record.api_port, + "status": inst.record.status, + **instance_warm_fields(request.app, inst.record)} + for inst in (_manager.instances() if _manager is not None else [])] + return web.json_response({ "node_id": config.node_id, "node_name": config.node_name, @@ -1400,6 +1414,7 @@ async def handle_status(request: web.Request) -> web.Response: "gpu": gpu_info, "engine_ready": engine_ready, "engine_adopted": engine_adopted, + "instances": instances, # Coarse engine load phase for the UI launching card (3c), derived from # the live /v1/models probe above rather than the engine's own latch: # see engine_load_phase. diff --git a/ainode/api/systemone.py b/ainode/api/systemone.py index de20553e..1589de6f 100644 --- a/ainode/api/systemone.py +++ b/ainode/api/systemone.py @@ -8,22 +8,26 @@ and no fork of the client. ONE adapter, so none of them needs an AINode-shaped branch. -What it does NOT promise: calibration. The hosted service's numbers are a +What it does NOT promise: the hosted service's calibration. Its numbers are a property of the model TypeSafe trained and of how they fit it; these are the probabilities the served model put on the option labels, read off its logprobs -and renormalized, and nothing here corrects, tempers or rescales them. A local -model's confidence is worth what that model's confidence is worth, and the way -to find out is ``scripts/ainode-bench.py decide``, which scores exactly this -against labels. So: same wire, same client, same typed answers, and no claim at -all about how well those answers are spread. +and renormalized. The one correction applied is the served model's OWN: a +decision adapter that ships ``temperatures.json`` beside its weights gets each +question's logprobs divided by the temperature fitted for that question's type, +in the decision core (``api/decide.py``), and the response's top-level +``calibration`` block says whether that happened and with which temperatures. A +request sends ``"calibration": "raw"`` to get the engine's own spread instead. A +local model's confidence is worth what that model's confidence is worth, and the +way to find out is ``scripts/ainode-bench.py decide``, which scores exactly this +against labels. Two numbers on the way out are inferences rather than readings, and each says so where it is computed: ``confidence`` is the hosted service's chance-corrected formula (``normalized_confidence``), reproduced from its published examples because no document states it, and the usage block falls back to the bench's own -estimate when an engine reports no usage at all (``usage_block``). The raw -distribution goes out beside them untouched, so a caller who disagrees with -either has the numbers they came from. +estimate when an engine reports no usage at all (``usage_block``). The +distribution goes out beside them unchanged by either, so a caller who disagrees +with either has the numbers they came from. One engine call answers one question, which is also what bounds a question: only the top ``MAX_CRITERIA`` labels come back with a probability, so a wider option @@ -61,8 +65,12 @@ from aiohttp import web from ainode.api.decide import ( + CHOICE, + NOUL, + SCORE, TOP_LOGPROBS, DecideError, + calibration_mode, candidates_for, merge_usage, normalize_questions, @@ -72,12 +80,10 @@ unavailable, ) -# The three question types the format defines. Anything else is a 422 rather than -# a guess: a client that asked for a kind of judgement this route does not have is -# better off being told which kinds it has. -CHOICE = "choice" -NOUL = "noul" -SCORE = "score" +# The three question types the format defines, which are also the kinds a decision +# adapter's temperatures are fitted per (``decide.py``). Anything else is a 422 +# rather than a guess: a client that asked for a kind of judgement this route does +# not have is better off being told which kinds it has. QUESTION_TYPES = (CHOICE, NOUL, SCORE) # A noul's two options, in this order, always. The answer is P(true), so `true` @@ -283,11 +289,15 @@ def decide_questions(translated: dict[str, Translated]) -> dict[str, dict]: Goes through the core's own ``normalize_questions`` rather than around it, so the option ceiling, the two-option floor and distinct options are checked in - one place for both routes. + one place for both routes. Each question keeps its Jev type as its ``kind``, + which is what picks the adapter's temperature for it. """ - return normalize_questions({key: {"question": item.question, - "options": item.options} - for key, item in translated.items()}) + questions = normalize_questions({key: {"question": item.question, + "options": item.options} + for key, item in translated.items()}) + for key, item in translated.items(): + questions[key]["kind"] = item.kind + return questions # ---------------------------------------------------------------- translate out @@ -322,9 +332,9 @@ def normalized_confidence(top: Any, options: int) -> float: INFERRED, not specified: it reproduces every example in TypeSafe's published docs and SDK types for both choice and score, and Kev's playground authors - arrived at the same formula for choice, but no document states it. The raw - distribution goes out untouched in ``probabilities`` beside it, so a caller - who disagrees with the formula has the numbers it came from. + arrived at the same formula for choice, but no document states it. The + distribution it came from goes out in ``probabilities`` beside it, so a + caller who disagrees with the formula has the numbers it came from. """ spread = probability(top) if options < 2: @@ -461,6 +471,7 @@ async def handle_systemone(request: web.Request) -> web.Response: translated = translate_questions(body.get("questions")) questions = decide_questions(translated) state = serialize_state(body.get("state")) + calibration = calibration_mode(body.get("calibration")) except DecideError as exc: return unprocessable(str(exc)) @@ -477,7 +488,8 @@ async def handle_systemone(request: web.Request) -> web.Response: # No shared instructions block: in this format a question's own instructions # are the whole prompt for it, and the questions of one ask still never see # each other's answers. - run = await run_questions(request, model, questions, state, None, candidates) + run = await run_questions(request, model, questions, state, None, candidates, + calibration) answers: dict[str, dict] = {} failures = list(run.failures) @@ -503,4 +515,5 @@ async def handle_systemone(request: web.Request) -> web.Response: "answers": answers, "usage": usage_block(run.payloads, state, translated, len(answers)), "latency_ms": round(total_ms, 1), + "calibration": run.calibration, }) diff --git a/ainode/models/api_routes.py b/ainode/models/api_routes.py index 36d66827..2acb5e45 100644 --- a/ainode/models/api_routes.py +++ b/ainode/models/api_routes.py @@ -1270,15 +1270,31 @@ async def _wait_for_bind(app, port: int, backend, timeout: float = 300.0, Every verdict this returns, bound or not, is appended to the node's launch-time ledger, the only place a load time is written down where the interface can read it back afterwards (see ``record_launch_time``). + + A bind is also where a decision model starts warming its answer grammar + (``api/decide.py::schedule_decision_warmup``, #277), in the background and + without holding the verdict: every launch path waits here, so no path can + skip it. """ bound, reason, alive = await _bind_wait(app, port, backend, timeout, loading=loading) await record_launch_time(app, port, backend, seconds=alive, outcome="ready" if bound else "failed", reason="" if bound else reason) + if bound: + _schedule_decision_warmup(app, port, backend) return bound, reason, alive +def _schedule_decision_warmup(app, port: int, backend) -> None: + """Start a decision model's warm-up. Never lets a failure reach the bind.""" + try: + from ainode.api.decide import schedule_decision_warmup + schedule_decision_warmup(app, port, backend) + except Exception: + logger.exception("could not schedule the decision warm-up on :%s", port) + + async def _bind_wait(app, port: int, backend, timeout: float = 300.0, loading: int = 1): """The wait itself. The contract, the signals and the verdicts are documented @@ -1796,6 +1812,9 @@ async def _await_primary_bind(app, config, loading: int = 1) -> bool: "container that outlived the restart (up %s)", adopted_primary.get("model") or model, port, adopted_primary.get("uptime") or "unknown") + # The warm flag died with the old process. An engine that already + # compiled answers the warm-up at once, and one that did not needs it. + _schedule_decision_warmup(app, port, boot_engine) return True # Otherwise wait for it to serve, retrying it once if it died on the way up: # a node that comes back with its main model silently missing is the #96 shape. diff --git a/tests/test_decide_calibration.py b/tests/test_decide_calibration.py new file mode 100644 index 00000000..7bb01fdc --- /dev/null +++ b/tests/test_decide_calibration.py @@ -0,0 +1,262 @@ +"""The adapter's own temperatures on /v1/decide and /v1/systemone (#276). + +A decision model's store directory can carry ``temperatures.json`` beside the +weights, one fitted temperature per question kind. When the served model's +directory has one, each question's label logprobs are divided by its kind's +temperature before the softmax; ``"calibration": "raw"`` opts out; and both +routes say what was applied. The route tests reuse the fake vLLM the decide tests +drive, with this node's model store pointed at a temp directory. +""" + +import json +import math + +import pytest +import pytest_asyncio +from aiohttp.test_utils import TestClient, TestServer + +from ainode.api.decide import ( + CHOICE, + NOUL, + SCORE, + DecideError, + calibration_for, + calibration_mode, + decision_from_payload, + distribution_from_logprobs, + is_decision_model_dir, + model_store_dir, + normalize_questions, + read_temperatures, +) +from tests.test_decide import MODEL, QUESTIONS, TICKET, FakeEngine, _app, _payload +from tests.test_systemone import QUESTIONS as JEV_QUESTIONS + +TEMPS = {"choice": 2.0, "noul": 0.5, "score": 1.5} + + +def _store(models_dir, temps=TEMPS, name="temperatures.json"): + """A model directory the way our downloader writes one: ``org--name``.""" + directory = models_dir / MODEL.replace("/", "--") + directory.mkdir(parents=True) + (directory / "config.json").write_text("{}") + if temps is not None: + (directory / name).write_text(json.dumps({"temperatures": temps, + "fitted_on": "held-out"})) + return directory + + +def _softmax(logprobs, temperature=1.0): + top = max(logprobs) + weights = [math.exp((lp - top) / temperature) for lp in logprobs] + return [w / sum(weights) for w in weights] + + +def _fake_logprobs(count): + """What ``FakeEngine`` puts on the labels when it picks the first one.""" + return [-0.0625] + [-4.0 - i for i in range(1, count)] + + +# ------------------------------------------------------------------- the math + + +def test_a_temperature_divides_the_logprobs_before_the_softmax(): + tops = [{"token": "A", "logprob": -0.1}, {"token": "B", "logprob": -2.1}] + raw = distribution_from_logprobs(["A", "B"], tops) + hot = distribution_from_logprobs(["A", "B"], tops, temperature=2.0) + cold = distribution_from_logprobs(["A", "B"], tops, temperature=0.5) + assert raw["A"] == pytest.approx(1 / (1 + math.exp(-2.0)), abs=1e-6) + assert hot["A"] == pytest.approx(1 / (1 + math.exp(-1.0)), abs=1e-6) + assert cold["A"] == pytest.approx(1 / (1 + math.exp(-4.0)), abs=1e-6) + # Tempering flattens or sharpens, never reorders. + assert cold["A"] > raw["A"] > hot["A"] > 0.5 + + +def test_a_decision_reads_its_confidence_off_the_tempered_spread(): + payload = _payload("A", [("A", -0.1), ("B", -2.1)]) + raw = decision_from_payload(payload, ["yes", "no"], 10.0) + hot = decision_from_payload(payload, ["yes", "no"], 10.0, temperature=2.0) + assert raw["answer"] == hot["answer"] == "yes" + assert hot["confidence"] == pytest.approx(1 / (1 + math.exp(-1.0)), abs=1e-6) + assert hot["confidence"] < raw["confidence"] + assert sum(hot["distribution"].values()) == pytest.approx(1.0, abs=1e-5) + + +def test_each_decide_question_is_tagged_with_the_kind_it_is_fitted_as(): + out = normalize_questions(QUESTIONS) + assert out["category"]["kind"] == CHOICE + assert out["urgency"]["kind"] == SCORE + assert out["needs_human"]["kind"] == NOUL + + +# ------------------------------------------------------------ reading the file + + +def test_the_temperatures_are_read_from_the_model_s_store_directory(tmp_path): + directory = _store(tmp_path) + assert model_store_dir(tmp_path, MODEL) == directory + assert is_decision_model_dir(directory) + assert read_temperatures(directory) == TEMPS + + +def test_a_prompt_contract_alone_marks_a_decision_model(tmp_path): + directory = _store(tmp_path, temps=None) + assert not is_decision_model_dir(directory) + (directory / "prompt_contract.json").write_text("{}") + assert is_decision_model_dir(directory) + assert read_temperatures(directory) is None + + +def test_a_bad_entry_leaves_that_kind_at_the_engine_s_own_spread(tmp_path): + directory = _store(tmp_path, temps={"choice": 0, "noul": "hot", "score": 1.25, + "other": 3.0, "extra": True}) + assert read_temperatures(directory) == {"score": 1.25} + (directory / "temperatures.json").write_text("not json") + assert read_temperatures(directory) is None + (directory / "temperatures.json").write_text(json.dumps({"temperatures": []})) + assert read_temperatures(directory) is None + + +def test_the_calibration_block_says_what_was_applied(tmp_path): + assert calibration_for(tmp_path, MODEL, None) == {"applied": False, + "temperatures": None} + _store(tmp_path) + assert calibration_for(tmp_path, MODEL, None) == {"applied": True, + "temperatures": TEMPS} + # Opting out still shows what the caller opted out of. + assert calibration_for(tmp_path, MODEL, "raw") == {"applied": False, + "temperatures": TEMPS} + + +def test_raw_is_the_only_calibration_a_request_can_ask_for(): + assert calibration_mode(None) is None + assert calibration_mode("raw") == "raw" + with pytest.raises(DecideError, match="'calibration' must be \"raw\""): + calibration_mode("tempered") + + +# ------------------------------------------------------------------ the routes + + +def _calibrated_app(port, models_dir): + app = _app(port) + app["config"].models_dir = str(models_dir) + return app + + +@pytest_asyncio.fixture +async def engine_server(): + fake = FakeEngine() + server = TestServer(fake.app()) + await server.start_server() + try: + yield server + finally: + await server.close() + + +@pytest_asyncio.fixture +async def calibrated(engine_server, tmp_path): + _store(tmp_path) + async with TestClient(TestServer(_calibrated_app(engine_server.port, + tmp_path))) as c: + yield c + + +@pytest_asyncio.fixture +async def uncalibrated(engine_server, tmp_path): + _store(tmp_path, temps=None) + async with TestClient(TestServer(_calibrated_app(engine_server.port, + tmp_path))) as c: + yield c + + +def _decide_body(**over): + body = {"model": MODEL, "state": TICKET, "questions": QUESTIONS} + body.update(over) + return body + + +@pytest.mark.asyncio +async def test_decide_applies_each_question_kind_s_temperature(calibrated): + resp = await calibrated.post("/v1/decide", json=_decide_body()) + assert resp.status == 200 + data = await resp.json() + assert data["calibration"] == {"applied": True, "temperatures": TEMPS} + decisions = data["decisions"] + cases = (("category", 5, TEMPS["choice"]), ("urgency", 5, TEMPS["score"]), + ("needs_human", 2, TEMPS["noul"])) + for key, count, temperature in cases: + expected = _softmax(_fake_logprobs(count), temperature) + assert decisions[key]["confidence"] == pytest.approx(expected[0], abs=1e-5), key + got = list(decisions[key]["distribution"].values()) + assert got == pytest.approx(expected, abs=1e-5), key + + +@pytest.mark.asyncio +async def test_decide_raw_opt_out_is_the_engine_s_own_spread(calibrated): + resp = await calibrated.post("/v1/decide", json=_decide_body(calibration="raw")) + assert resp.status == 200 + data = await resp.json() + assert data["calibration"] == {"applied": False, "temperatures": TEMPS} + expected = _softmax(_fake_logprobs(5)) + assert data["decisions"]["category"]["confidence"] == pytest.approx(expected[0], + abs=1e-5) + + +@pytest.mark.asyncio +async def test_decide_without_a_temperatures_file_is_unchanged(uncalibrated): + resp = await uncalibrated.post("/v1/decide", json=_decide_body()) + assert resp.status == 200 + data = await resp.json() + assert data["calibration"] == {"applied": False, "temperatures": None} + expected = _softmax(_fake_logprobs(2)) + assert data["decisions"]["needs_human"]["confidence"] == pytest.approx( + expected[0], abs=1e-5) + + +@pytest.mark.asyncio +async def test_decide_refuses_an_unknown_calibration_with_a_400(calibrated): + resp = await calibrated.post("/v1/decide", json=_decide_body(calibration="warm")) + assert resp.status == 400 + assert "'calibration'" in (await resp.json())["error"]["message"] + + +def _jev_body(**over): + body = {"model": MODEL, "state": TICKET, "questions": JEV_QUESTIONS} + body.update(over) + return body + + +@pytest.mark.asyncio +async def test_systemone_applies_the_temperatures_and_reports_them(calibrated): + resp = await calibrated.post("/v1/systemone", json=_jev_body()) + assert resp.status == 200 + data = await resp.json() + assert data["calibration"] == {"applied": True, "temperatures": TEMPS} + answers = data["answers"] + queue = _softmax(_fake_logprobs(3), TEMPS["choice"]) + assert answers["queue"]["probabilities"]["billing"] == pytest.approx(queue[0], + abs=1e-5) + # Chance-corrected off the TEMPERED top probability. + assert answers["queue"]["confidence"] == pytest.approx((3 * queue[0] - 1) / 2, + abs=1e-5) + noul = _softmax(_fake_logprobs(2), TEMPS["noul"]) + assert answers["needs_human"]["noul"] == pytest.approx(noul[0], abs=1e-5) + + +@pytest.mark.asyncio +async def test_systemone_raw_opt_out_reports_temperatures_it_did_not_apply(calibrated): + resp = await calibrated.post("/v1/systemone", json=_jev_body(calibration="raw")) + assert resp.status == 200 + data = await resp.json() + assert data["calibration"] == {"applied": False, "temperatures": TEMPS} + noul = _softmax(_fake_logprobs(2)) + assert data["answers"]["needs_human"]["noul"] == pytest.approx(noul[0], abs=1e-5) + + +@pytest.mark.asyncio +async def test_systemone_refuses_an_unknown_calibration_with_a_422(calibrated): + resp = await calibrated.post("/v1/systemone", json=_jev_body(calibration=1)) + assert resp.status == 422 + assert "'calibration'" in (await resp.json())["error"]["message"] diff --git a/tests/test_decide_warmup.py b/tests/test_decide_warmup.py new file mode 100644 index 00000000..80827b18 --- /dev/null +++ b/tests/test_decide_warmup.py @@ -0,0 +1,191 @@ +"""Warming a decision engine's answer grammar on bind (#277). + +The first grammar-constrained request per question shape costs a cold compile, +so a decision model (a store directory with ``prompt_contract.json`` or +``temperatures.json``) is sent one minimal question per kind as soon as its +engine binds, and ``/api/status`` reports ``warm`` per instance. A fake vLLM +records what the warm-up actually sent. +""" + +import asyncio +import json +import socket +from types import SimpleNamespace + +import aiohttp +import pytest +import pytest_asyncio +from aiohttp.test_utils import TestClient, TestServer + +from ainode.api import decide +from ainode.api.decide import ( + ask_one, + schedule_decision_warmup, + warm_decision_engine, +) +from ainode.api.server import create_app +from ainode.core.config import NodeConfig +from ainode.discovery.instance import InstanceRecord +from ainode.engine.instance_manager import InstanceManager +from ainode.models import api_routes +from tests.test_decide import MODEL, FakeEngine + + +@pytest_asyncio.fixture +async def fake(): + engine = FakeEngine() + server = TestServer(engine.app()) + await server.start_server() + try: + yield engine, server.port + finally: + await server.close() + + +def _decision_store(models_dir, marker="temperatures.json"): + directory = models_dir / MODEL.replace("/", "--") + directory.mkdir(parents=True) + (directory / "config.json").write_text("{}") + (directory / marker).write_text(json.dumps({"temperatures": {"choice": 1.2}})) + return directory + + +def _backend(models_dir, served=None): + return SimpleNamespace(config=SimpleNamespace(model=MODEL, models_dir=str(models_dir), + served_model_name=served)) + + +async def _drain(): + await asyncio.gather(*list(decide._WARM_TASKS)) + + +@pytest.mark.asyncio +async def test_the_warm_up_sends_one_constrained_request_per_kind(fake): + engine, port = fake + status: dict = {} + async with aiohttp.ClientSession() as session: + assert await warm_decision_engine(session, port, MODEL, status) + assert len(engine.seen) == 3 + for body in engine.seen: + # The same body the routes send: constrained, logprobs on, thinking off. + assert body["model"] == MODEL + assert body["structured_outputs"] == {"choice": ["A", "B"]} + assert body["logprobs"] is True + assert body["chat_template_kwargs"]["enable_thinking"] is False + noul = engine.seen[1]["messages"][-1]["content"] + assert "A. true" in noul and "B. false" in noul + assert status["warm"] is True and status["warming"] is False + assert list(status["compile_seconds"]) == ["choice", "noul", "score"] + assert status["error"] is None + + +@pytest.mark.asyncio +async def test_a_warm_up_that_fails_says_where_and_stays_cold(): + engine = FakeEngine(status=500) + server = TestServer(engine.app()) + await server.start_server() + try: + status: dict = {} + async with aiohttp.ClientSession() as session: + assert not await warm_decision_engine(session, server.port, MODEL, status) + finally: + await server.close() + assert status["warm"] is False + assert status["error"].startswith("choice:") + assert len(engine.seen) == 1, "the first failure stops the warm-up" + + +@pytest.mark.asyncio +async def test_a_bind_warms_a_decision_model_and_records_the_flag(fake, tmp_path, + monkeypatch): + engine, port = fake + _decision_store(tmp_path, marker="prompt_contract.json") + + async def bound(*_a, **_k): + return True, "bound", 1.0 + + async def no_ledger(*_a, **_k): + return None + + monkeypatch.setattr(api_routes, "_bind_wait", bound) + monkeypatch.setattr(api_routes, "record_launch_time", no_ledger) + app = {"decision_warm": {}, "client_session": None} + ok, _, _ = await api_routes._wait_for_bind(app, port, _backend(tmp_path)) + assert ok + assert app["decision_warm"][port]["warm"] is False, "warming, not yet warm" + await _drain() + assert app["decision_warm"][port]["warm"] is True + assert len(engine.seen) == 3 + + +@pytest.mark.asyncio +async def test_a_model_that_is_not_a_decision_model_is_left_alone(fake, tmp_path): + engine, port = fake + (tmp_path / MODEL.replace("/", "--")).mkdir() + (tmp_path / MODEL.replace("/", "--") / "config.json").write_text("{}") + app = {"decision_warm": {port: {"model": "old", "warm": True}}} + assert not schedule_decision_warmup(app, port, _backend(tmp_path)) + assert port not in app["decision_warm"], "a stale entry on the port is cleared" + assert engine.seen == [] + + +@pytest.mark.asyncio +async def test_the_warm_up_asks_by_the_served_model_name(fake, tmp_path): + engine, port = fake + _decision_store(tmp_path) + app = {"decision_warm": {}, "client_session": None} + assert schedule_decision_warmup(app, port, _backend(tmp_path, served=["judge"])) + await _drain() + assert {body["model"] for body in engine.seen} == {"judge"} + + +@pytest.mark.asyncio +async def test_a_route_timeout_names_the_grammar_compile_not_an_unreachable_node(): + class Slow(FakeEngine): + async def completions(self, request): + await asyncio.sleep(1.0) + return await super().completions(request) + + server = TestServer(Slow().app()) + await server.start_server() + try: + async with aiohttp.ClientSession() as session: + payload, _, err = await ask_one(session, [("127.0.0.1", server.port)], + {"model": MODEL}, timeout_s=0.1) + finally: + await server.close() + assert payload is None + assert "compiling the answer grammar, retry" in err + assert "unreachable" not in err + + +def _free_port(): + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@pytest.mark.asyncio +async def test_status_reports_warm_per_instance(tmp_path): + config = NodeConfig(node_id="n1", node_name="TestNode", model=None, + api_port=_free_port(), web_port=_free_port(), + models_dir=str(tmp_path / "models")) + app = create_app(config=config, engine=None) + manager = InstanceManager(base_port=config.api_port) + manager.add(InstanceRecord(instance_id="n1:judge", model=MODEL, api_port=8001, + status="serving"), None) + manager.add(InstanceRecord(instance_id="n1:chat", model="org/chat", api_port=8002, + status="serving"), None) + app["instances"] = manager + app["decision_warm"][8001] = {"model": MODEL, "warm": False, + "compile_seconds": {"choice": 71.2}} + async with TestClient(TestServer(app)) as client: + rows = {row["api_port"]: row + for row in (await (await client.get("/api/status")).json())["instances"]} + assert rows[8001]["warm"] is False + assert rows[8001]["warm_compile_seconds"] == {"choice": 71.2} + assert rows[8002]["warm"] is None, "nothing to warm on a chat model" + app["decision_warm"][8001]["warm"] = True + rows = {row["api_port"]: row + for row in (await (await client.get("/api/status")).json())["instances"]} + assert rows[8001]["warm"] is True diff --git a/tests/test_systemone.py b/tests/test_systemone.py index d15aea4b..163fa53c 100644 --- a/tests/test_systemone.py +++ b/tests/test_systemone.py @@ -452,7 +452,9 @@ async def test_every_question_type_round_trips(client, engine_fake): data = await resp.json() assert data["model"] == MODEL assert data["latency_ms"] > 0 - assert set(data) == {"model", "answers", "usage", "latency_ms"} + assert set(data) == {"model", "answers", "usage", "latency_ms", "calibration"} + # No temperatures.json in this node's store: the engine's own spread. + assert data["calibration"] == {"applied": False, "temperatures": None} assert list(data["answers"]) == ["queue", "needs_human", "severity"] assert parse_answers_like_jde(data["answers"]) is not None