Blame for source/tests/test_server.py

"""Contract tests for the HTTP API (deg.api / deg.server / deg.client).

Golden fixtures:
* ``fixtures/openai_decisions_docs.json``: the OpenAI Decisions guide's requests and
  illustrative responses (openai-python 3.26.1 types);
* ``fixtures/jev_systemone_live.json``: live api.typesafe.ai (jev-1.13.0) requests,
  statuses and responses.

The model is :class:`fake_fill.FakeFill` (no torch); responses still go through
``deg.response.assemble`` and its schema validation.
"""

from __future__ import annotations

import json
import threading
import time
from pathlib import Path
from typing import Any

import pytest
from fake_fill import FakeFill

from deg.api import parse_decisions, parse_systemone
from deg.client import Deg, DegAPIError
from deg.response import MODEL_ID, confidence
from deg.server import Batcher, Engine, make_server

FIXTURES = Path(__file__).parent / "fixtures"
DOCS = json.loads((FIXTURES / "openai_decisions_docs.json").read_text())["cases"]
JEV = json.loads((FIXTURES / "jev_systemone_live.json").read_text())["cases"]

EVENT_INPUT = ("Hi, order AB-1234 arrived with a cracked screen on 2026-10-01. "
               "I want a refund, not a replacement.")
EVENT_QUESTIONS = [
    {"type": "choice", "name": "event", "instructions": "Which event should be dispatched?",
     "choices": [{"value": "refund", "description": "Customer asks for money back."},
                 {"value": "replacement", "description": "Customer wants a new unit."},
                 {"value": "other"}]},
    {"type": "predicate", "name": "damaged", "instructions": "cracked or broken item?"},
    {"type": "extract", "name": "order_id", "instructions": "The order id."},
    {"type": "extract", "name": "delivered", "instructions": "Delivery date.",
     "value_type": "date"},
    {"type": "extract", "name": "tracking", "instructions": "Tracking number.",
     "pattern": "^1Z\\w+$"},
    {"type": "generate", "name": "note", "instructions": "A one-line note for the agent.",
     "max_length": 3, "max_tokens": 64},
]


class Server:
    def __init__(self, *, api_key: str | None = None, fill: Any = None,
                 fill_many: Any = None, window_s: float = 0.001, max_pending: int = 256):
        self.fill = fill or FakeFill()
        self.batcher = Batcher(self.fill, fill_many=fill_many, window_s=window_s,
                               max_pending=max_pending)
        self.httpd = make_server(Engine(self.batcher), "127.0.0.1", 0, api_key=api_key)
        self.url = f"http://127.0.0.1:{self.httpd.server_address[1]}"
        threading.Thread(target=self.httpd.serve_forever, daemon=True).start()

    def close(self) -> None:
        self.httpd.shutdown()
        self.httpd.server_close()


@pytest.fixture(scope="module")
def server():
    srv = Server()
    yield srv
    srv.close()


@pytest.fixture(scope="module")
def client(server):
    return Deg(server.url)


def shape(value: Any) -> Any:
    """Keys (in order) and JSON types, recursively; lists by their first item."""
    if isinstance(value, dict):
        return {k: shape(v) for k, v in value.items()}
    if isinstance(value, list):
        return [shape(value[0])] if value else []
    if isinstance(value, bool):
        return "bool"
    if isinstance(value, (int, float)):
        return "number"
    return type(value).__name__


def raw_post(url: str, path: str, body: Any, headers: dict | None = None) -> tuple[int, Any]:
    import urllib.error
    import urllib.request

    data = body if isinstance(body, bytes) else json.dumps(body).encode()
    req = urllib.request.Request(url + path, data=data, method="POST",
                                 headers={"Content-Type": "application/json", **(headers or {})})
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return r.status, json.loads(r.read())
    except urllib.error.HTTPError as exc:
        return exc.code, json.loads(exc.read())


# ---------------------------------------------------------------- OpenAI Decisions


@pytest.mark.parametrize("case", ["choice", "score"])
def test_decisions_docs_golden_shape(client, case):
    golden = DOCS[case]
    got = client.decisions.create(**golden["request"])
    excerpt = golden["response_excerpt"]
    assert shape(got["answers"]) == shape(excerpt["answers"])
    assert list(got) == ["answers", "model", "usage"]
    assert got.model == MODEL_ID
    usage = got.usage
    assert list(usage) == ["input_tokens", "input_tokens_details", "output_tokens",
                           "output_tokens_details", "total_tokens"]
    assert usage.total_tokens == usage.input_tokens + usage.output_tokens > 0
    assert usage.output_tokens == 0  # decisions generate nothing
    ans = got.answers[0]
    probs = [p.probability for p in ans.probabilities]
    assert abs(sum(probs) - 1) < 1e-9
    assert ans.confidence == pytest.approx(confidence(probs))
    if case == "choice":
        assert [p.value for p in ans.probabilities] == [
            c["value"] for c in golden["request"]["questions"][0]["choices"]]
        assert ans.choice == max(ans.probabilities, key=lambda p: p.probability).value
    else:
        assert [(p.value, p.label) for p in ans.probabilities] == [
            (0, "Cosmetic"), (1, "Workaround available"), (2, "Fully blocked")]
        assert ans.score == pytest.approx(sum(p.value * p.probability
                                              for p in ans.probabilities))


def test_decisions_docs_score_formula():
    """The guide's own numbers: probabilities .1/.7/.2 over indices 0/1/2 -> score 1.1."""
    from deg.api import _expected

    assert _expected([0.1, 0.7, 0.2]) == pytest.approx(1.1)


def test_decisions_predicate_text_and_messages(client):
    req = json.loads(json.dumps(DOCS["predicate_image"]["request"]))
    req["input"][0]["content"] = [{"type": "input_text", "text": "Does the box show damage?"}]
    got = client.decisions.create(**req)
    assert shape(got.answers) == shape(DOCS["predicate_image"]["response_excerpt"]["answers"])
    assert 0 <= got.answers[0].probability <= 1


def test_decisions_image_refused(server):
    status, body = raw_post(server.url, "/v1/decisions", DOCS["predicate_image"]["request"])
    assert status == 400
    assert body["error"]["type"] == "invalid_request_error"
    assert "input_image is not supported" in body["error"]["message"]


def test_decisions_extensions_event_dispatch(server, client):
    got = client.decisions.create(model="deg-latest", input=EVENT_INPUT,
                                  questions=EVENT_QUESTIONS)
    a = {x.name: x for x in got.answers}
    assert [x.name for x in got.answers] == [q["name"] for q in EVENT_QUESTIONS]
    assert a["event"].type == "choice" and a["event"].choice == "refund"
    assert a["damaged"].type == "predicate"
    assert a["order_id"] == {"name": "order_id", "type": "extract", "value": "AB-1234",
                             "span": [10, 17], "confidence": 0.9}
    assert EVENT_INPUT[10:17] == "AB-1234"
    assert a["delivered"].value == "2026-10-01"
    assert a["tracking"].value is None and a["tracking"].null_reason == "not_found"
    assert a["tracking"].span is None
    assert a["note"] == {"name": "note", "type": "generate", "text": "not",
                         "output_tokens": 2}
    assert got.usage.output_tokens == 2
    # one fill for the whole request; extensions compiled to the documented schema
    _, schema, options = server.fill.calls[-1]
    assert list(schema["properties"]) == [q["name"] for q in EVENT_QUESTIONS]
    assert schema["properties"]["note"] == {"type": "string", "x-generate": True,
                                            "description": "A one-line note for the agent.",
                                            "maxLength": 3, "x-max-tokens": 64}
    assert schema["properties"]["delivered"]["format"] == "date"
    assert schema["properties"]["order_id"]["type"] == ["string", "null"]
    assert options.max_new_tokens == 256  # request default; the note has its own budget


def test_decisions_score_levels_go_to_x_levels(server, client):
    got = client.decisions.create(model="deg-latest", input="I am furious", questions=[
        {"type": "score", "name": "anger", "instructions": "How angry is the customer?",
         "levels": [{"label": "calm", "description": "no complaint"}, {"label": "annoyed"},
                    {"label": "furious", "description": "threatens to leave"}]}])
    _, schema, _ = server.fill.calls[-1]
    assert schema["properties"]["anger"] == {
        "type": "integer", "minimum": 0, "maximum": 2,
        "description": "How angry is the customer?",
        "x-levels": {"0": "calm: no complaint", "1": "annoyed",
                     "2": "furious: threatens to leave"}}
    assert [p.label for p in got.answers[0].probabilities] == ["calm", "annoyed", "furious"]


def test_generate_questions_mix_max_tokens(server, client):
    got = client.decisions.create(model="deg-latest", input="refund please", questions=[
        {"type": "generate", "name": "a", "instructions": "x", "max_tokens": 8},
        {"type": "generate", "name": "b", "instructions": "y", "max_tokens": 9},
        {"type": "generate", "name": "c", "instructions": "z"}])
    assert [a.type for a in got.answers] == ["generate"] * 3
    _, schema, options = server.fill.calls[-1]
    props = schema["properties"]
    assert props["a"]["x-max-tokens"] == 8 and props["b"]["x-max-tokens"] == 9
    assert "x-max-tokens" not in props["c"] and options.max_new_tokens == 256
    status, _ = raw_post(server.url, "/v1/systemone", {
        "model": "jev-latest", "state": "refund please", "questions": {
            "a": {"type": "generate", "instructions": "x", "max_tokens": 4},
            "b": {"type": "generate", "instructions": "y", "max_tokens": 12}}})
    assert status == 200
    _, schema, _ = server.fill.calls[-1]
    assert [schema["properties"][k]["x-max-tokens"] for k in "ab"] == [4, 12]


def test_decisions_unnamed_questions_get_null_name(client):
    got = client.decisions.create(model="deg-latest", input="refund please", questions=[
        {"type": "predicate", "instructions": "refund requested?"},
        {"type": "predicate", "instructions": "angry?"}])
    assert [a.name for a in got.answers] == [None, None]


def test_decisions_single_choice_and_level_without_model(server, client):
    before = len(server.fill.calls)
    got = client.decisions.create(model="deg-latest", input="x", questions=[
        {"type": "choice", "name": "c", "instructions": "pick", "choices": [{"value": True}]},
        {"type": "score", "name": "s", "instructions": "rate", "levels": [{"label": "only"}]}])
    assert len(server.fill.calls) == before
    assert got.answers[0].choice is True and got.answers[0].confidence == 1.0
    assert got.answers[1].score == 0 and got.usage.input_tokens == 0


@pytest.mark.parametrize("questions,needle", [
    ([{"type": "classify", "instructions": "x"}], "type must be one of"),
    ([{"type": "choice", "instructions": "x",
       "choices": [{"value": f"o{i}"} for i in range(256)]}], "Too many choices"),
    ([{"type": "score", "instructions": "x",
       "levels": [{"label": str(i)} for i in range(11)]}], "Too many score levels"),
    ([{"type": "score", "instructions": "x", "levels": []}], "levels must be a non-empty"),
    ([{"type": "choice", "instructions": "x", "choices": []}], "choices must be a non-empty"),
    ([{"type": "choice", "instructions": "x",
       "choices": [{"value": "a"}, {"value": "a"}]}], "duplicate choice value"),
    ([{"type": "predicate", "name": "a", "instructions": "x"},
      {"type": "predicate", "name": "a", "instructions": "y"}], "names must be unique"),
    ([{"type": "predicate", "instructions": "x", "choices": []}], "Unrecognized question"),
    ([{"type": "generate", "instructions": "x", "max_tokens": 0}], "max_tokens must be"),
    ([{"type": "generate", "instructions": "x", "pattern": "("}], "pattern"),
    ([{"type": "extract", "instructions": "x", "value_type": "date",
       "max_length": 3}], "applies only to string"),
    ([{"type": "extract", "instructions": "x", "value_type": "phone"}], "value_type"),
    ([], "non-empty array"),
])
def test_decisions_validation_errors(server, questions, needle):
    status, body = raw_post(server.url, "/v1/decisions",
                            {"model": "gpt-6-luna", "input": "x", "questions": questions})
    assert status == 400, body
    assert set(body["error"]) == {"message", "type", "param", "code"}
    assert needle in body["error"]["message"]


def test_decisions_choice_values_are_typed(client):
    got = client.decisions.create(model="deg-latest", input="true", questions=[
        {"type": "choice", "name": "c", "instructions": "pick",
         "choices": [{"value": "true"}, {"value": True}]}])
    assert [p.value for p in got.answers[0].probabilities] == ["true", True]


def test_decisions_unknown_model_and_fields(server):
    status, body = raw_post(server.url, "/v1/decisions", {
        "model": "gpt-5", "input": "x", "questions": [{"type": "predicate",
                                                       "instructions": "y"}], "foo": 1})
    assert status == 400
    msg = body["error"]["message"]
    assert "Unrecognized request argument supplied: foo" in msg
    assert "The model `gpt-5` does not exist" in msg


def test_decisions_request_options(server, client):
    with pytest.raises(DegAPIError) as exc:
        client.decisions.create(model="deg-latest", input="x", options={"temperature": 0.5},
                                questions=[{"type": "generate", "instructions": "y"}])
    assert exc.value.status == 400 and "requires a seed" in exc.value.body["error"]["message"]
    client.decisions.create(model="deg-latest", input="x",
                            options={"temperature": 0.5, "seed": 3},
                            questions=[{"type": "generate", "instructions": "y"}])
    assert server.fill.calls[-1][2].seed == 3


# ---------------------------------------------------------------- stock openai SDK


def test_stock_openai_sdk_round_trip(server):
    openai = pytest.importorskip("openai")
    from openai.types.decision import (
        AnswerAnswerResourceChoice,
        AnswerAnswerResourcePredicate,
        AnswerAnswerResourceScore,
        Decision,
    )

    oai = openai.OpenAI(base_url=server.url + "/v1", api_key="unused", max_retries=0)
    for case in ("choice", "score"):
        got = oai.decisions.create(**DOCS[case]["request"])
        assert isinstance(got, Decision)
        want = {"choice": AnswerAnswerResourceChoice, "score": AnswerAnswerResourceScore}[case]
        assert isinstance(got.answers[0], want)
        assert got.answers[0].name == DOCS[case]["request"]["questions"][0]["name"]
        assert got.usage.total_tokens > 0
    got = oai.decisions.create(model="gpt-6-luna", input=[
        {"role": "user", "content": [{"type": "input_text", "text": "cracked lid"}]}],
        questions=[{"type": "predicate", "name": "visible_damage",
                    "instructions": "cracked anything?"}])
    assert isinstance(got.answers[0], AnswerAnswerResourcePredicate)
    assert got.answers[0].probability == pytest.approx(0.8)


def test_stock_openai_sdk_extensions_and_errors(server):
    openai = pytest.importorskip("openai")
    oai = openai.OpenAI(base_url=server.url + "/v1", api_key="unused", max_retries=0)
    got = oai.decisions.create(model="deg-latest", input=EVENT_INPUT,
                               questions=EVENT_QUESTIONS,
                               extra_body={"options": {"date_order": "MDY"}})
    by_name = {a.name: a for a in got.answers}
    assert by_name["event"].choice == "refund"
    # extension answers are not in the SDK's union; their fields stay readable
    assert by_name["order_id"].type == "extract"
    assert by_name["order_id"].value == "AB-1234"
    assert by_name["note"].text == "not"
    assert server.fill.calls[-1][2].date_order == "MDY"
    with pytest.raises(openai.BadRequestError) as exc:
        oai.decisions.create(model="gpt-6-luna", input="x",
                             questions=[{"type": "score", "instructions": "y",
                                         "levels": [{"label": str(i)} for i in range(11)]}])
    assert "Too many score levels" in str(exc.value)


def test_stock_openai_sdk_auth():
    openai = pytest.importorskip("openai")
    srv = Server(api_key="k1")
    try:
        ok = openai.OpenAI(base_url=srv.url + "/v1", api_key="k1", max_retries=0)
        assert ok.decisions.create(**DOCS["choice"]["request"]).answers[0].type == "choice"
        bad = openai.OpenAI(base_url=srv.url + "/v1", api_key="nope", max_retries=0)
        with pytest.raises(openai.AuthenticationError):
            bad.decisions.create(**DOCS["choice"]["request"])
        status, body = raw_post(srv.url, "/v1/systemone", JEV["ok_all"]["request"])
        assert status == 403
        assert body == {"detail": {"error_type": "authentication_error", "message":
                                   "Must supply an API key! Check your request and try again."}}
    finally:
        srv.close()


# ---------------------------------------------------------------- Jev SystemOne


@pytest.mark.parametrize("case", ["ok_all", "uncertain", "obj_state", "extra_q_name",
                                  "one_option", "one_level"])
def test_systemone_live_shapes(server, case):
    live = JEV[case]
    status, body = raw_post(server.url, "/v1/systemone", live["request"])
    assert status == live["status"] == 200
    assert shape(body) == shape(live["response"])  # keys in the same order, same types
    assert list(body["answers"]) == list(live["request"]["questions"])
    for qid, ans in body["answers"].items():
        if ans["type"] in ("choice", "score"):
            probs = list(ans["probabilities"].values())
            assert abs(sum(probs) - 1) < 1e-9
            assert ans["confidence"] == pytest.approx(confidence(probs) if len(probs) > 1
                                                      else 1.0)
        if ans["type"] == "score":
            assert ans["legend"] == live["response"]["answers"][qid]["legend"]
            assert ans["score"] == pytest.approx(sum(int(k) * v for k, v in
                                                     ans["probabilities"].items()))
    assert body["model"] == MODEL_ID


@pytest.mark.parametrize("case", ["unknown_type", "extra_top", "bad_model", "opts256",
                                  "eleven_levels", "no_model", "no_questions"])
def test_systemone_live_errors(server, case):
    live = JEV[case]
    request = json.loads(json.dumps(live["request"]))
    # deg accepts two things live Jev rejects (opt-in extensions): the question type
    # `extract` and the top-level `options`; probe the same rule with other names.
    if case == "unknown_type":
        request["questions"]["x"]["type"] = "classify"
    if case == "extra_top":
        request["foo"] = request.pop("options")
    status, body = raw_post(server.url, "/v1/systemone", request)
    assert status == live["status"]
    if case == "bad_model":
        assert body == live["response"]
    elif status == 400:
        assert body == live["response"]
    else:  # 422: the same pydantic-style item (type, loc, msg)
        got = [{k: d[k] for k in ("type", "loc", "msg")} for d in body["detail"]]
        want = [{k: d[k] for k in ("type", "loc", "msg")} for d in live["response"]["detail"]]
        assert got == want


def test_systemone_noul_criteria_and_score_levels_in_schema(server):
    raw_post(server.url, "/v1/systemone", JEV["ok_all"]["request"])
    _, schema, _ = server.fill.calls[-1]
    props = schema["properties"]
    assert props["urgent_crit"]["description"] == (
        "Is the customer asking for urgent action?\nYes means: Asks for action now\n"
        "No means: No time pressure")
    assert props["anger"] == {"type": "integer", "minimum": 0, "maximum": 3,
                              "description": "How angry is the customer?",
                              "x-levels": {"0": "calm", "1": "annoyed", "2": "angry",
                                           "3": "furious"}}
    assert props["department"]["oneOf"][2] == {"const": "shipping"}


def test_systemone_extensions(client):
    got = client.systemone(EVENT_INPUT, {
        "event": {"type": "choice", "instructions": "Which event?",
                  "criteria": {"refund": "money back", "replacement": None}},
        "order_id": {"type": "extract", "instructions": "The order id."},
        "note": {"type": "generate", "instructions": "Note for the agent.", "max_length": 4}})
    assert got.answers["event"].choice == "refund"
    assert got.answers["order_id"] == {"type": "extract", "value": "AB-1234", "span": [10, 17],
                                       "confidence": 0.9}
    assert got.answers["note"] == {"type": "generate", "text": "note", "output_tokens": 2}
    assert got.usage == {"input_tokens": got.usage.input_tokens, "output_tokens": 2}


def test_systemone_existing_jev_client_unchanged(server):
    """The Clef/SystemOne client (rwkv_omarchy.rl.clef_server.ClefJudge, model
    "clef-flash") works against deg as-is."""
    clef = pytest.importorskip("rwkv_omarchy.rl.clef_server")
    QUESTIONS, ClefJudge, build_request = clef.QUESTIONS, clef.ClefJudge, clef.build_request

    judge = ClefJudge(server.url)
    out = judge.judge_request(build_request("fix dns", [{"command": "ls", "output": "x"}],
                                            "Done, DNS fixed."))
    assert set(out) >= set(QUESTIONS) and out["input_tokens"] > 0


# ---------------------------------------------------------------- /v1/fill


def test_fill_round_trip_and_x_generate_on_enum(client):
    schema = {"type": "object", "properties": {
        "event": {"enum": ["refund", "replacement"]},
        "order_id": {"type": "string"},
        "note": {"type": "string", "x-generate": True, "maxLength": 4}},
        "required": ["event", "order_id", "note"]}
    got = client.fill(EVENT_INPUT, schema)
    assert got.value == {"event": "refund", "order_id": "AB-1234", "note": "note"}
    assert got.meta["/order_id"].span == [10, 17]
    assert got.usage.generated_tokens == 2
    bad = json.loads(json.dumps(schema))
    bad["properties"]["event"]["x-generate"] = True
    with pytest.raises(DegAPIError) as exc:
        client.fill(EVENT_INPUT, bad)
    assert exc.value.status == 422
    assert exc.value.body["error"]["type"] == "invalid_request"
    assert any("x-generate" in d["msg"] for d in exc.value.body["error"]["detail"])


def test_fill_unknown_option_is_422(client):
    with pytest.raises(DegAPIError) as exc:
        client.fill("x", {"type": "object", "properties": {"a": {"type": "boolean"}}},
                    options={"beam": 4})
    assert exc.value.status == 422


def test_malformed_json_bodies(server):
    assert raw_post(server.url, "/v1/decisions", b"{")[0] == 400
    assert raw_post(server.url, "/v1/systemone", b"{")[0] == 422
    assert raw_post(server.url, "/v1/fill", b"{")[0] == 422
    assert raw_post(server.url, "/v1/nope", {})[0] == 404


def test_models_and_health(client):
    ids = [m.id for m in client.models().data]
    assert {"deg-latest", MODEL_ID, "gpt-6-luna", "jev-latest"} <= set(ids)
    health = client.health()
    assert health.ok and health.model == MODEL_ID and health.batcher["jobs"] >= 0


# ---------------------------------------------------------------- batching


def test_batcher_groups_concurrent_jobs_and_isolates_failures():
    seen: list[int] = []
    fake = FakeFill()

    def fill(ctx, schema, opts):
        if ctx == "bad":
            raise ValueError("boom")
        return fake(ctx, schema, opts)

    def fill_many(jobs):
        seen.append(len(jobs))
        time.sleep(0.05)
        return [ValueError("boom") if ctx == "bad" else fake(ctx, schema, opts)
                for ctx, schema, opts in jobs]

    srv = Server(fill=fill, fill_many=fill_many, window_s=0.2)
    try:
        req = {"model": "deg-latest", "questions": [{"type": "predicate",
                                                       "instructions": "refund?"}]}
        results: dict[int, tuple[int, Any]] = {}

        def go(i: int) -> None:
            results[i] = raw_post(srv.url, "/v1/decisions",
                                  dict(req, input="bad" if i == 3 else f"refund {i}"))

        threads = [threading.Thread(target=go, args=(i,)) for i in range(6)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        assert max(seen) > 1
        assert srv.batcher.stats()["max_seen_batch"] > 1
        assert results[3][0] == 500 and results[3][1]["error"]["type"] == "server_error"
        assert all(results[i][0] == 200 for i in range(6) if i != 3)
    finally:
        srv.close()


def test_server_fill_many_equals_sequential_fill():
    """The real runtime bound like ``--ckpt`` (``bind_model``) on a tiny random Qwen3.5:
    concurrent requests batched through ``fill_many`` answer exactly as one-at-a-time
    ``fill`` (SPEC.md §10)."""
    torch = pytest.importorskip("torch")
    pytest.importorskip("transformers")
    from test_model import SNAPSHOT, tiny_lm

    from deg.model import load_model, load_tokenizer
    from deg.server import bind_model

    if SNAPSHOT is None:
        pytest.skip("Qwen3.5-0.8B tokenizer not cached")
    model = load_model(None, device="cpu", dtype=torch.float32, lm=tiny_lm(),
                       tokenizer=load_tokenizer(SNAPSHOT))
    torch.manual_seed(1)
    with torch.no_grad():
        for p in model.heads.parameters():
            p.normal_(0.0, 0.5)
    fill, fill_many = bind_model(model)
    assert fill_many is not None
    calls: list[int] = []

    def counted_many(jobs):
        calls.append(len(jobs))
        return fill_many(jobs)

    reqs = [
        ("/v1/decisions", {"model": "deg-latest", "input": EVENT_INPUT, "questions": [
            *EVENT_QUESTIONS[:4],
            {"type": "score", "name": "anger", "instructions": "How angry?",
             "levels": [{"label": "calm"}, {"label": "angry", "description": "upset"}]},
            {"type": "generate", "name": "n1", "instructions": "Short note.", "max_tokens": 3},
            {"type": "generate", "name": "n2", "instructions": "Reply.", "max_tokens": 5}]}),
        ("/v1/systemone", {"model": "jev-latest", "state": {"msg": "late parcel, angry"},
                           "questions": {"anger": {"type": "score", "instructions": "anger",
                                                   "criteria": ["calm", "angry", "furious"]},
                                         "late": {"type": "noul", "instructions": "late?"}}}),
        ("/v1/decisions", {"model": "deg-latest", "input": "Order AB-9 for $80.",
                           "questions": [EVENT_QUESTIONS[0], EVENT_QUESTIONS[2]]}),
    ]
    seq = Server(fill=fill, fill_many=None)
    try:
        want = [raw_post(seq.url, path, body) for path, body in reqs]
    finally:
        seq.close()
    assert all(status == 200 for status, _ in want), want
    srv = Server(fill=fill, fill_many=counted_many, window_s=0.5)
    try:
        got: dict[int, Any] = {}
        threads = [threading.Thread(target=lambda i=i: got.__setitem__(i, raw_post(
            srv.url, *reqs[i]))) for i in range(len(reqs))]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
    finally:
        srv.close()
    assert calls and max(calls) > 1
    assert [got[i] for i in range(len(reqs))] == want


def test_batcher_rejects_when_full():
    gate = threading.Event()

    def slow(ctx, schema, opts):
        gate.wait(5)
        return FakeFill()(ctx, schema, opts)

    srv = Server(fill=slow, max_pending=1)
    try:
        req = {"model": "deg-latest", "input": "x",
               "questions": [{"type": "predicate", "instructions": "y"}]}
        out: list = []
        threads = [threading.Thread(target=lambda: out.append(
            raw_post(srv.url, "/v1/decisions", req))) for _ in range(4)]
        for t in threads:
            t.start()
        time.sleep(0.5)
        gate.set()
        for t in threads:
            t.join()
        codes = sorted(c for c, _ in out)
        assert 429 in codes and 200 in codes
        body = next(b for c, b in out if c == 429)
        assert body["error"]["code"] == "rate_limit_exceeded"
    finally:
        srv.close()


# ---------------------------------------------------------------- pure parsing


def test_parse_is_shared_between_apis():
    """predicate == noul, choice == choice, score == score: same field schemas."""
    o = parse_decisions({"model": "deg-latest", "input": "s", "questions": [
        {"type": "predicate", "name": "a", "instructions": "i"},
        {"type": "choice", "name": "b", "instructions": "i",
         "choices": [{"value": "x", "description": "dx"}, {"value": "y"}]},
        {"type": "score", "name": "c", "instructions": "i",
         "levels": [{"label": "lo"}, {"label": "hi"}]}]})
    j = parse_systemone({"model": "deg-latest", "state": "s", "questions": {
        "a": {"type": "noul", "instructions": "i"},
        "b": {"type": "choice", "instructions": "i", "criteria": {"x": "dx", "y": None}},
        "c": {"type": "score", "instructions": "i", "criteria": ["lo", "hi"]}}})
    assert o.schema == j.schema