[Engine] API fields accepted and silently not honoured: one list to pick from, most with no known client #4839

# usage, from the repo root: PYTHONPATH=. python fields_mock.py    (mock engine only; no model, no GPU)
import http.client, json, sys, tempfile
from pathlib import Path
from serve.frontend import ChatTemplate
from serve.server import ByteTokenizer, MockEngine, Service, serve
from serve.test_logprobs import LpEngine
import serve.images as images_api, serve.test_images as TI

ONE = "</think>\n\n<tool_call>\n<function=a>\n<parameter=x>\nnot-int\n</parameter>\n</function>\n</tool_call>"
TWO = ONE + "\n<tool_call>\n<function=b>\n<parameter=y>\n2\n</parameter>\n</function>\n</tool_call>"
P = {"type": "object", "properties": {"x": {"type": "integer"}}, "required": ["x"]}
OAI = [{"type": "function", "function": {"name": n, "parameters": P}} for n in "ab"]
RSP = [{"type": "function", "name": n, "parameters": P} for n in "ab"]
ANT = [{"name": n, "input_schema": P} for n in "ab"]
U = [{"role": "user", "content": "hi"}]

def up(script="</think>\n\nanswer", cls=MockEngine):
    tok = ByteTokenizer()
    eng = cls(tok, script) if cls is LpEngine else cls(tok, script, max_context=131072)
    svc = Service(eng, tok, ChatTemplate(Path("serve/chat_template.jinja")))
    return svc, eng, tok, serve(svc, port=0).server_address[1]

def call(port, path, body=None, method="POST"):
    c = http.client.HTTPConnection("127.0.0.1", port, timeout=30)
    c.request(method, path, None if body is None else json.dumps(dict(model="m", **body)), {"Content-Type": "application/json"})
    r = c.getresponse(); raw = r.read().decode(errors="replace")
    try: return r.status, r.getheader("Content-Type"), json.loads(raw)
    except ValueError: return r.status, r.getheader("Content-Type"), raw

def chat(port, **kw): return call(port, "/v1/chat/completions", dict(messages=U, **kw))[2]["choices"]
def names(c): return [t["function"]["name"] for t in c[0]["message"].get("tool_calls") or []]
def anth(port, **kw): return call(port, "/v1/messages", dict(dict(messages=U, max_tokens=300), **kw))[2]
def rsp(port, **kw): return call(port, "/v1/responses", dict(input="hi", **kw))[2]

_, e, t, p = up(TWO)
print("parallel_tool_calls false, chat:", names(chat(p, tools=OAI, parallel_tool_calls=False)))
print("parallel_tool_calls false, responses:", [o["name"] for o in rsp(p, tools=RSP, parallel_tool_calls=False)["output"]])
print("disable_parallel_tool_use:", [b["name"] for b in anth(p, tools=ANT, tool_choice={"type": "auto", "disable_parallel_tool_use": True})["content"]])
print("allowed_tools [b]:", names(chat(p, tools=OAI, tool_choice={"type": "allowed_tools", "allowed_tools": {"mode": "auto", "tools": [{"type": "function", "function": {"name": "b"}}]}})))
print("n=3 choices:", len(chat(p, n=3)), "| n=0 status:", call(p, "/v1/chat/completions", dict(messages=U, n=0))[0])

_, e, t, p = up(ONE)
strict = [{"type": "function", "function": {"name": "a", "strict": True, "parameters": P}}]
print("strict tool args:", chat(p, tools=strict)[0]["message"]["tool_calls"][0]["function"]["arguments"])
r = chat(p, functions=[{"name": "a", "parameters": P}], function_call={"name": "a"})[0]
print("functions + function_call:", r["finish_reason"], "| function_call:", r["message"].get("function_call"), "| tool_calls:", r["message"].get("tool_calls"), "| in prompt:", '"a"' in t.decode(e.last_prompt))
print("tool_choice none, responses:", [o["type"] for o in rsp(p, tools=RSP, tool_choice="none")["output"]])
print("tool_choice none, chat:", chat(p, tools=OAI, tool_choice="none")[0]["finish_reason"])

_, e, t, p = up("secret</think>\n\nanswer")
anth(p, messages=[U[0], {"role": "assistant", "content": "The answer is ("}])
print("prefill, prompt ends:", repr(t.decode(e.last_prompt)[-48:]))
anth(p, tools=[{"type": "web_search_20250305", "name": "web_search", "max_uses": 5}])
print("typed tool, in prompt:", '{"name": "web_search", "description": "", "parameters": {}}' in t.decode(e.last_prompt))
hist = lambda **kw: [U[0], {"role": "assistant", "content": [{"type": "tool_use", "id": "t1", "name": "a", "input": {}}]},
                     {"role": "user", "content": [dict(type="tool_result", tool_use_id="t1", content="boom", **kw)]}]
anth(p, tools=ANT, messages=hist(is_error=True)); bad = t.decode(e.last_prompt)
anth(p, tools=ANT, messages=hist()); print("is_error, same prompt as success:", bad == t.decode(e.last_prompt))
print("thinking display omitted:", anth(p, thinking={"type": "enabled", "budget_tokens": 2000, "display": "omitted"})["content"][0])
anth(p, thinking={"type": "between_tools"}); print("between_tools, prompt ends:", repr(t.decode(e.last_prompt)[-10:]))
s = call(p, "/v1/messages/count_tokens", dict(messages=U, tools=ANT, tool_choice={"type": "none"}))[2]["input_tokens"]
u = anth(p, tools=ANT, tool_choice={"type": "none"})["usage"]
print("count_tokens none:", s, "| usage none:", u["input_tokens"] + u.get("cache_creation_input_tokens", 0) + u.get("cache_read_input_tokens", 0))
print("/v1/models entry:", call(p, "/v1/models", method="GET")[2]["data"][0].keys())
print("responses usage details:", rsp(p)["usage"]["input_tokens_details"])
print("input_tokens:", call(p, "/v1/responses/input_tokens", dict(input="hi"))[0::2])
print("HEAD /health:", call(p, "/health", method="HEAD")[:2])
print("DELETE /v1/responses/x:", call(p, "/v1/responses/x", method="DELETE")[:2])

_, e, t, p = up("</think>\n\nanswer", LpEngine)
print("responses logprobs:", [c.get("logprobs") for o in rsp(p, top_logprobs=2, include=["message.output_text.logprobs"])["output"] if o["type"] == "message" for c in o["content"]])
print("chat logprobs entries:", len(chat(p, logprobs=True, top_logprobs=2)[0]["logprobs"]["content"]))

_, e, t, p = up('</think>\n\n{"a":"bad"}')
fmt = {"type": "json_schema", "json_schema": {"name": "s", "strict": True, "schema": {"type": "object", "properties": {"a": {"type": "integer"}}, "required": ["a"]}}}
print("json_schema answer of the wrong type:", call(p, "/v1/chat/completions", dict(messages=U, response_format=fmt))[0])

svc, e, t, p = up()
real = images_api.sd_cli_args
images_api.sd_cli_args = lambda cfg, job, out: [sys.executable, str(TI.FAKE), *real(cfg, job, out)[1:]]
svc.images = images_api.ImageGen(TI.image_cfg(tempfile.mkdtemp(), keep_files=3))
r = call(p, "/v1/images/generations", dict(prompt="a fox", stream=True))
print("images stream:", r[0], r[1], list(r[2]) if isinstance(r[2], dict) else r[2][:80])