"""``POST /v1/fill`` request contract (SPEC.md §5). Parsing only; no model runtime yet.""" from __future__ import annotations import json import math from dataclasses import dataclass from typing import Any from deg.parsers import DateOrder from deg.schema import FillPlan, SchemaError, compile_schema MODEL_ALIASES = frozenset({"deg-latest", "deg-qwen3.5-0.8b-0.1"}) OPTION_KEYS = frozenset( {"max_new_tokens", "max_items", "date_order", "default_offset", "temperature", "seed", "array_decode", "null_margin"} ) # how an anchored array's items are found (SPEC.md §12.6): threshold = anchor p > 0.5; # count_peaks = the count head's N, then the N best anchor peaks; hybrid = threshold, and # count_peaks when no anchor fires but the count head says N > 0 ARRAY_DECODES = ("threshold", "count_peaks", "hybrid") class RequestError(ValueError): """Client error (HTTP 422) with a list of ``{loc, msg}``.""" def __init__(self, errors: list[dict[str, Any]], status: int = 422) -> None: super().__init__("; ".join(f"{'/'.join(map(str, e['loc']))}: {e['msg']}" for e in errors)) self.errors = errors self.status = status def body(self) -> dict[str, Any]: return {"error": {"type": "invalid_request", "status": self.status, "detail": self.errors}} @dataclass(frozen=True) class FillOptions: max_new_tokens: int = 256 # per generate field, on top of maxLength # array length cap when the schema gives no maxItems; None = no cap (anchored arrays # are unbounded; index-addressed arrays stop at the count head's MAX_COUNT) max_items: int | None = None date_order: DateOrder | None = None # numeric a/b/yyyy dates; None = only unambiguous default_offset: str = "Z" # date-time spans without an offset temperature: float = 0.0 # 0 = greedy (default); > 0 requires seed seed: int | None = None # one of ARRAY_DECODES; count_peaks once the count head is trained by anchor_count # (run-2b step-4500: results/deg/regress-2b; hybrid was picked at step-3000 before it) array_decode: str = "count_peaks" # extract null bias in nats (SPEC.md §12.1): a nullable/optional extract leaf is null # when its best valid span's log-prob < null log-prob + null_margin. > 0 trades # made-up answers for omissions. None = the server's default (--null-margin), else 0 null_margin: float | None = None @dataclass(frozen=True) class FillRequest: context: Any context_text: str plan: FillPlan options: FillOptions model: str def render_context(context: Any) -> str: """The exact text the model reads once; span offsets index into it (code points).""" if isinstance(context, str): return context return json.dumps(context, ensure_ascii=False, separators=(", ", ": "), allow_nan=False) def parse_options(raw: Any, errors: list[dict[str, Any]]) -> FillOptions: loc = ["body", "options"] if raw is None: return FillOptions() if not isinstance(raw, dict): errors.append({"loc": loc, "msg": "options must be an object"}) return FillOptions() for key in raw: if key not in OPTION_KEYS: errors.append({"loc": [*loc, key], "msg": "unknown option"}) values = FillOptions().__dict__ | {k: v for k, v in raw.items() if k in OPTION_KEYS} for key in ("max_new_tokens", "max_items"): v = values[key] if key == "max_items" and v is None: continue if not isinstance(v, int) or isinstance(v, bool) or v < 1: errors.append({"loc": [*loc, key], "msg": "must be a positive integer"}) if values["date_order"] not in (None, "MDY", "DMY"): errors.append({"loc": [*loc, "date_order"], "msg": "must be null, 'MDY' or 'DMY'"}) offset = values["default_offset"] if not (isinstance(offset, str) and (offset == "Z" or ( len(offset) == 6 and offset[0] in "+-" and offset[3] == ":" and offset[1:3].isdigit() and offset[4:].isdigit()))): errors.append({"loc": [*loc, "default_offset"], "msg": "must be 'Z' or ±HH:MM"}) if values["array_decode"] not in ARRAY_DECODES: errors.append({"loc": [*loc, "array_decode"], "msg": "must be one of " + ", ".join(repr(x) for x in ARRAY_DECODES)}) margin = values["null_margin"] if margin is not None and not (isinstance(margin, (int, float)) and not isinstance( margin, bool) and math.isfinite(margin)): errors.append({"loc": [*loc, "null_margin"], "msg": "must be null or a finite number"}) temp = values["temperature"] if not (isinstance(temp, (int, float)) and not isinstance(temp, bool) and math.isfinite(temp) and temp >= 0): errors.append({"loc": [*loc, "temperature"], "msg": "must be a finite number >= 0"}) seed = values["seed"] if seed is not None and (not isinstance(seed, int) or isinstance(seed, bool)): errors.append({"loc": [*loc, "seed"], "msg": "must be an integer"}) elif isinstance(temp, (int, float)) and temp > 0 and seed is None: errors.append({"loc": [*loc, "seed"], "msg": "temperature > 0 requires a seed"}) try: return FillOptions(**values) except TypeError: return FillOptions() def parse_fill_request(payload: Any) -> FillRequest: """Validate ``{context, schema, model?, options?}``. Raises :class:`RequestError`.""" if not isinstance(payload, dict): raise RequestError([{"loc": ["body"], "msg": "request body must be a JSON object"}]) errors: list[dict[str, Any]] = [] for key in payload: if key not in {"context", "schema", "model", "options"}: errors.append({"loc": ["body", key], "msg": "unknown field"}) context = payload.get("context") context_text = "" if context is None: errors.append({"loc": ["body", "context"], "msg": "field required"}) else: try: context_text = render_context(context) except (TypeError, ValueError): errors.append({"loc": ["body", "context"], "msg": "context must be finite JSON"}) if not context_text.strip() or context_text in ("{}", "[]"): errors.append({"loc": ["body", "context"], "msg": "context must not be empty"}) model = payload.get("model", "deg-latest") if model not in MODEL_ALIASES: errors.append({"loc": ["body", "model"], "msg": f"unknown model {model!r}"}) options = parse_options(payload.get("options"), errors) plan = None if "schema" not in payload: errors.append({"loc": ["body", "schema"], "msg": "field required"}) else: try: plan = compile_schema(payload["schema"]) except SchemaError as exc: errors.extend({"loc": ["body", *e["loc"]], "msg": e["msg"]} for e in exc.errors) if errors: raise RequestError(errors) assert plan is not None return FillRequest(context, context_text, plan, options, model) def generate_arg(spec: str) -> str: """argparse type of ``--generate``: ``self``, ``base`` (the two-copy hybrid), ``base+lora:`` (the hybrid with a GENERATE LoRA merged into the base copy; :func:`deg.model.parse_generate` checks the directory at load) or ``moe`` (a deg-MoE checkpoint's W experts; its default).""" import argparse if spec in ("self", "base", "moe") or (spec.startswith("base+lora:") and len(spec) > 10): return spec raise argparse.ArgumentTypeError(f"expected self, base, moe or base+lora:: {spec!r}")