--enable-auto-tool-choice
--tool-call-parser qwen3_coder
--structured-outputs-config '{"backend": "xgrammar", "disable_any_whitespace": true}'
--reasoning-parser qwen3
--reasoning-config '{"reasoning_start_str": "<think>", "reasoning_end_str": "I have to give the solution based on the reasoning directly now.</think>"}'
--enable-prefix-caching
--enable-chunked-prefill
--max-num-batched-tokens 32768 --max-num-seqs 32
--override-generation-config '{"temperature": 1.0, "top_p": 0.95, "top_k": 20, "min_p": 0.0, "presence_penalty": 1.5}'

---

export VLLM_BASE_URL=https://your-vllm/v1
export VLLM_API_KEY=...
export VLLM_MODEL=Qwen/Qwen3.6-35B-A3B-FP8
python repro.py 20

---

== responses: {'valid': 20} channels={'reasoning': 20}
== chat:      {'valid': 20} channels={'content': 20}

---

{"output": [{"type": "reasoning", "...": "the JSON answer lives here"}],
   "output_text": null}

---

#!/usr/bin/env python3
"""Minimal repro: Responses API + json_schema structured output is NOT grammar-
constrained on a server with --reasoning-parser, while Chat Completions IS.

Same model, same schema, same prompt, thinking disabled on both paths.
- /v1/responses + text.format=json_schema   -> output lands in an unconstrained
  `reasoning` item; xgrammar never binds; Qwen intermittently runs away
  (bracket-spam `}]}]}...` -> invalid JSON).
- /v1/chat/completions + response_format=json_schema -> output stays in the
  grammar-constrained `content` channel; always valid.

Run:
    export VLLM_BASE_URL=https://your-vllm/v1
    export VLLM_API_KEY=...           # if your server requires it
    export VLLM_MODEL=Qwen/Qwen3.6-35B-A3B-FP8
    python repro.py [N]               # N iterations per path, default 20

Only dependency: httpx.
"""

import json
import os
import sys

import httpx

BASE_URL = os.environ["VLLM_BASE_URL"].rstrip("/")
API_KEY = os.environ.get("VLLM_API_KEY", "")
MODEL = os.environ.get("VLLM_MODEL", "Qwen/Qwen3.6-35B-A3B-FP8")
N = int(sys.argv[1]) if len(sys.argv) > 1 else 20

HEADERS = {"Content-Type": "application/json"}
if API_KEY:
    HEADERS["Authorization"] = f"Bearer {API_KEY}"

# A faithful-but-small extraction schema (list of objects), like the real
# workload that triggered the runaway. Bounded list, free-text strings.
SCHEMA = {
    "type": "object",
    "properties": {
        "offerings": {
            "type": "array",
            "maxItems": 8,
            "items": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "category": {"type": "string"},
                    "description": {"type": "string"},
                },
                "required": ["name", "category", "description"],
                "additionalProperties": False,
            },
        }
    },
    "required": ["offerings"],
    "additionalProperties": False,
}

SYSTEM = (
    "You extract a company's product/service offerings from the text and return "
    "them as structured data per the provided JSON schema. Be exhaustive."
)
# Long-ish, real-world-shaped input. The runaway is far more likely on a faithful
# prompt than on a toy one ("answer/score") — keep this representative.
USER = (
    "Company profile:\n"
    "Acme Logistics GmbH is a mid-size 3PL provider. We operate temperature-"
    "controlled warehousing across three sites, run a national same-day courier "
    "fleet, and offer customs brokerage for EU/non-EU shipments. Our digital arm "
    "ships a SaaS track-and-trace portal with a REST API, plus a managed EDI "
    "onboarding service for retail partners. We also do reverse-logistics / "
    "returns handling and ad-hoc project freight (oversized, hazmat). Recently we "
    "added a carbon-reporting dashboard for shippers.\n\n"
    "Extract the structured data per the schema."
)

TIMEOUT = httpx.Timeout(120.0)


def classify(text: str) -> str:
    """valid | empty | invalid-json | schema-mismatch"""
    text = (text or "").strip()
    if not text:
        return "empty"
    try:
        obj = json.loads(text)
    except json.JSONDecodeError:
        return "invalid-json"
    if not isinstance(obj, dict) or "offerings" not in obj:
        return "schema-mismatch"
    return "valid"


def run_responses(client: httpx.Client) -> tuple[str, str]:
    """Returns (classification, channel). channel = where the JSON text landed:
    'message' (grammar-constrained) vs 'reasoning' (unconstrained)."""
    body = {
        "model": MODEL,
        "input": [
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": USER},
        ],
        # reasoning omitted on purpose: our vLLM thinking-patch maps that to
        # enable_thinking=False. On stock vLLM, add the equivalent below.
        "text": {
            "format": {
                "type": "json_schema",
                "name": "offerings",
                "schema": SCHEMA,
                "strict": True,
            }
        },
        "max_output_tokens": 4096,
    }
    r = client.post(f"{BASE_URL}/responses", headers=HEADERS, json=body)
    r.raise_for_status()
    data = r.json()

    # Find where the text actually came out.
    msg_text, reasoning_text = "", ""
    for item in data.get("output", []):
        itype = item.get("type")
        parts = item.get("content") or []
        text = "".join(
            p.get("text", "") for p in parts if isinstance(p, dict)
        )
        if itype == "message":
            msg_text += text
        elif itype == "reasoning":
            reasoning_text += text + "".join(
                p.get("text", "") for p in [item] if "text" in item
            )
    # vLLM also exposes top-level output_text aggregating message items.
    output_text = data.get("output_text") or msg_text
    if output_text.strip():
        return classify(output_text), "message"
    if reasoning_text.strip():
        return classify(reasoning_text), "reasoning"
    return "empty", "none"


def run_chat(client: httpx.Client) -> tuple[str, str]:
    body = {
        "model": MODEL,
        "messages": [
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": USER},
        ],
        "response_format": {
            "type": "json_schema",
            "json_schema": {"name": "offerings", "schema": SCHEMA, "strict": True},
        },
        "max_tokens": 4096,
        # thinking patch does NOT hook the chat path -> disable explicitly.
        "chat_template_kwargs": {"enable_thinking": False},
    }
    r = client.post(f"{BASE_URL}/chat/completions", headers=HEADERS, json=body)
    r.raise_for_status()
    data = r.json()
    content = data["choices"][0]["message"].get("content") or ""
    return classify(content), "content"


def main() -> None:
    print(f"model={MODEL} base={BASE_URL} N={N}\n")
    with httpx.Client(timeout=TIMEOUT) as client:
        for label, fn in (("responses", run_responses), ("chat", run_chat)):
            counts: dict[str, int] = {}
            channels: dict[str, int] = {}
            for i in range(N):
                try:
                    cls, ch = fn(client)
                except Exception as e:  # noqa: BLE001 - repro, surface everything
                    cls, ch = f"http-error:{type(e).__name__}", "none"
                counts[cls] = counts.get(cls, 0) + 1
                channels[ch] = channels.get(ch, 0) + 1
                print(f"  {label:9} {i + 1:>3}: {cls:16} channel={ch}")
            print(f"== {label}: {dict(counts)} channels={dict(channels)}\n")


if __name__ == "__main__":
    main()