source /home/claude/venv/bin/activate          # or your own venv with requirements.txt installed
cd supportdesk
export PYTHONPATH=.
python -c "import pydantic, openai, httpx; print(pydantic.VERSION, openai.__version__, httpx.__version__)"

"""Why free text breaks downstream code: a real router fed three kinds of model output."""
from __future__ import annotations

import json
import re
import traceback

# The support desk's router: which queue a ticket goes to and how fast it must be answered.
QUEUES = {"billing": "finance-queue", "cancellation": "retention-queue", "account_access": "access-queue",
          "bug": "engineering-queue", "how_to": "tier1-queue", "feature_request": "product-queue"}
SLA_HOURS = {"urgent": 1, "high": 4, "normal": 24, "low": 72}


def route(triage: dict) -> str:
    """Downstream code written by someone who expects clean data."""
    queue = QUEUES[triage["category"]]
    hours = SLA_HOURS[triage["priority"]]
    owner = "human" if triage["needs_human"] else "assistant"
    return f"{queue} within {hours}h, handled by {owner}"


FREE_TEXT = """Category: Billing
Priority: High (money is wrong)
This customer needs a human because they want a refund."""

LOOKS_LIKE_JSON = """Sure! Here is the triage:
{"category": "billing", "priority": "high", "needs_human": true}"""

VALID = '{"category": "billing", "priority": "high", "needs_human": true}'


def try_route(label: str, text: str) -> None:
    print(f"--- {label}")
    try:
        print("OK:", route(json.loads(text)))
    except Exception as exc:  # show the failure the way a log would
        print("CRASH:", "".join(traceback.format_exception_only(exc)).strip())


try_route("valid JSON", VALID)
try_route("JSON with a friendly preamble", LOOKS_LIKE_JSON)
try_route("free text", FREE_TEXT)

# The usual "quick fix": scrape the free text with a regex. It runs, then fails later.
print("--- free text scraped with a regex")
fields = {k.lower(): v.strip() for k, v in re.findall(r"^(\w+):\s*(.+)$", FREE_TEXT, re.MULTILINE)}
print("scraped:", fields)
try:
    print("OK:", route(fields))
except Exception as exc:
    print("CRASH:", "".join(traceback.format_exception_only(exc)).strip())

--- valid JSON
OK: finance-queue within 4h, handled by human
--- JSON with a friendly preamble
CRASH: json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
--- free text
CRASH: json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
--- free text scraped with a regex
scraped: {'category': 'Billing', 'priority': 'High (money is wrong)'}
CRASH: KeyError: 'Billing'

"""Typed output schemas for the support assistant, validated with pydantic."""
from __future__ import annotations

from typing import Literal

from pydantic import BaseModel, ConfigDict, Field

Category = Literal["billing", "cancellation", "account_access", "bug", "how_to", "feature_request"]
Priority = Literal["low", "normal", "high", "urgent"]


class Triage(BaseModel):
    """What the assistant decides about one incoming ticket."""

    model_config = ConfigDict(extra="forbid")

    category: Category = Field(description="The single best category for the ticket.")
    priority: Priority = Field(description="urgent: many users blocked or security risk now; high: one user blocked or money wrong; normal: needs an answer; low: question or idea.")
    language: str = Field(pattern=r"^[a-z]{2}$", description="ISO 639-1 code of the customer's language, for example 'en'.")
    summary: str = Field(min_length=5, max_length=200, description="One sentence in English describing the request.")
    needs_human: bool = Field(description="True if an agent must act (refund, account change, legal, security) or the help center cannot answer it.")


class DraftReply(BaseModel):
    """A reply the assistant proposes; a human agent reviews it before sending."""

    model_config = ConfigDict(extra="forbid")

    reply: str = Field(min_length=1, max_length=2000)
    cited_articles: list[str] = Field(default_factory=list, max_length=3, description="Help-center article ids the reply relies on.")
    confidence: Literal["low", "medium", "high"]


def triage_json_schema() -> dict:
    """The JSON Schema for Triage, ready for a provider's structured-output mode."""
    return Triage.model_json_schema()

python -c "import json; from supportdesk.schemas import triage_json_schema; print(json.dumps(triage_json_schema(), indent=2))"

{
  "additionalProperties": false,
  "description": "What the assistant decides about one incoming ticket.",
  "properties": {
    "category": {
      "description": "The single best category for the ticket.",
      "enum": [
        "billing",
        "cancellation",
        "account_access",
        "bug",
        "how_to",
        "feature_request"
      ],
      "title": "Category",
      "type": "string"
    },
    "priority": {
      "description": "urgent: many users blocked or security risk now; high: one user blocked or money wrong; normal: needs an answer; low: question or idea.",
      "enum": [
        "low",
        "normal",
        "high",
        "urgent"
      ],
      "title": "Priority",
      "type": "string"
    },
    "language": {
      "description": "ISO 639-1 code of the customer's language, for example 'en'.",
      "pattern": "^[a-z]{2}$",
      "title": "Language",
      "type": "string"
    },
    "summary": {
      "description": "One sentence in English describing the request.",
      "maxLength": 200,
      "minLength": 5,
      "title": "Summary",
      "type": "string"
    },
    "needs_human": {
      "description": "True if an agent must act (refund, account change, legal, security) or the help center cannot answer it.",
      "title": "Needs Human",
      "type": "boolean"
    }
  },
  "required": [
    "category",
    "priority",
    "language",
    "summary",
    "needs_human"
  ],
  "title": "Triage",
  "type": "object"
}

"""Look at the schemas the assistant uses, and turn one into a strict response_format."""
from __future__ import annotations

import json

from examples.m06_structured import response_format_for, validate_output
from supportdesk.schemas import DraftReply, Triage, triage_json_schema

schema = triage_json_schema()
print("Triage required:", schema["required"])
print("Triage enums:", {k: v["enum"] for k, v in schema["properties"].items() if "enum" in v})
print("Triage bounds:", {k: {b: v[b] for b in ("pattern", "minLength", "maxLength") if b in v}
                         for k, v in schema["properties"].items() if any(b in v for b in ("pattern", "minLength"))})

print("\nDraftReply required (pydantic):", DraftReply.model_json_schema()["required"])
fmt = response_format_for(DraftReply)
print("DraftReply response_format for strict mode:")
print(json.dumps(fmt, indent=2))

# A strict-mode reply must include every field, so optional ones arrive as null.
from_model = {"reply": "You can export a board from Board menu > Export.", "cited_articles": None, "confidence": "high"}
print("\nvalidated:", validate_output(from_model, DraftReply))

"""Module 6 helpers: turn model text into validated objects, or fail loudly.

Four layers, each usable on its own:
  response_format_for  build a json_schema response_format for a provider
  parse_json           a tolerant parser for hostile model output
  generate_structured  call, parse, validate, and repair with an attempt ceiling
  salvage              keep the fields that are valid when the whole object is not
"""
from __future__ import annotations

import copy
import json
import re
from dataclasses import dataclass, field
from typing import Any, Callable

from pydantic import BaseModel, ValidationError

from supportdesk.llm import ChatResult, Usage

# ---------------------------------------------------------------------------
# 1. Native structured output: a json_schema response_format
# ---------------------------------------------------------------------------


def strict_schema(model_cls: type[BaseModel]) -> dict[str, Any]:
    """The model's JSON Schema, reshaped for strict mode.

    Strict mode (Groq documents it for gpt-oss) needs every property listed in
    `required` and `additionalProperties: false` on every object. A field that
    was optional becomes nullable instead: the model must write it, but may
    write null. `validate_output` turns those nulls back into defaults.
    """
    schema = copy.deepcopy(model_cls.model_json_schema())

    def fix(node: Any) -> None:
        if isinstance(node, dict):
            if isinstance(node.get("title"), str):      # a schema title, not a property named "title"
                node.pop("title")
            if node.get("type") == "object" and "properties" in node:
                was_required = set(node.get("required", []))
                for name, prop in node["properties"].items():
                    if name not in was_required:
                        prop.pop("default", None)
                        wrapped: dict[str, Any] = {"anyOf": [prop, {"type": "null"}]}
                        if "description" in prop:            # keep the description where the model reads it
                            wrapped = {"description": prop.pop("description"), **wrapped}
                        node["properties"][name] = wrapped
                node["required"] = list(node["properties"])
                node["additionalProperties"] = False
            for value in node.values():
                fix(value)
        elif isinstance(node, list):
            for value in node:
                fix(value)

    fix(schema)
    return schema


def response_format_for(model_cls: type[BaseModel], strict: bool = True) -> dict[str, Any]:
    """The `response_format` value for llm.chat(..., response_format=...)."""
    return {
        "type": "json_schema",
        "json_schema": {"name": model_cls.__name__, "schema": strict_schema(model_cls), "strict": strict},
    }


# ---------------------------------------------------------------------------
# 2. A tolerant parser, built in stages so each stage can be measured
# ---------------------------------------------------------------------------

FENCE = re.compile(r"```(?:json|JSON)?\s*\n?(.*?)(?:```|$)", re.DOTALL)
SMART_QUOTES = str.maketrans({"\u201c": '"', "\u201d": '"', "\u2018": "'", "\u2019": "'"})


class ParseError(ValueError):
    """Raised when no JSON object can be recovered from the text."""


@dataclass
class Parsed:
    value: dict[str, Any]
    repairs: list[str] = field(default_factory=list)   # what the parser had to fix, for logging


def strip_fences(text: str) -> tuple[str, bool]:
    """Return the inside of the first ```json fence, if there is one."""
    match = FENCE.search(text)
    return (match.group(1), True) if match else (text, False)


def extract_object(text: str) -> tuple[str, bool]:
    """Cut from the first '{' to its matching '}', skipping braces inside strings.

    If the text ends before the object closes, return everything from '{' on:
    the truncation stage may still be able to close it.
    """
    start = text.find("{")
    if start == -1:
        raise ParseError("no '{' in output")
    depth, in_str, quote, escaped = 0, False, "", False
    for i in range(start, len(text)):
        ch = text[i]
        if in_str:
            if escaped:
                escaped = False
            elif ch == "\\":
                escaped = True
            elif ch == quote:
                in_str = False
        elif ch in "\"'":
            in_str, quote = True, ch
        elif ch == "{":
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0:
                return text[start:i + 1], (start > 0 or i + 1 < len(text.rstrip()))
    return text[start:], start > 0


def normalize(text: str) -> tuple[str, list[str]]:
    """Fix the JSON-adjacent syntax models write: single quotes, trailing commas,
    Python literals, and smart quotes. Works token by token, so text inside
    double-quoted strings is never touched."""
    repairs: list[str] = []
    if any(c in text for c in "\u201c\u201d\u2018\u2019"):
        text = text.translate(SMART_QUOTES)
        repairs.append("smart_quotes")
    out: list[str] = []
    i, n = 0, len(text)
    while i < n:
        ch = text[i]
        if ch == '"':                                   # copy a JSON string unchanged
            j = i + 1
            while j < n and text[j] != '"':
                j += 2 if text[j] == "\\" else 1
            out.append(text[i:j + 1])
            i = j + 1
        elif ch == "'":                                 # 'single quoted' -> "double quoted"
            j = i + 1
            while j < n and text[j] != "'":
                j += 2 if text[j] == "\\" else 1
            out.append(json.dumps(text[i + 1:j].replace("\\'", "'")))
            if "single_quotes" not in repairs:
                repairs.append("single_quotes")
            i = j + 1
        elif ch == ",":                                 # drop a comma that closes nothing
            j = i + 1
            while j < n and text[j] in " \t\r\n":
                j += 1
            if j < n and text[j] in "}]":
                if "trailing_comma" not in repairs:
                    repairs.append("trailing_comma")
            else:
                out.append(ch)
            i += 1
        elif text.startswith(("True", "False", "None"), i) and not (i and text[i - 1].isalnum()):
            word = next(w for w in ("True", "False", "None") if text.startswith(w, i))
            out.append({"True": "true", "False": "false", "None": "null"}[word])
            if "python_literals" not in repairs:
                repairs.append("python_literals")
            i += len(word)
        else:
            out.append(ch)
            i += 1
    return "".join(out), repairs


def close_truncated(text: str) -> tuple[str, bool]:
    """Close an object that was cut off mid-stream (for example by max_tokens).

    Closes an open string, drops a dangling key or comma, then appends the
    closing brackets in the right order. The value may be incomplete, so the
    caller must treat a closed object as suspect.
    """
    stack: list[str] = []
    in_str, escaped = False, False
    for ch in text:
        if in_str:
            if escaped:
                escaped = False
            elif ch == "\\":
                escaped = True
            elif ch == '"':
                in_str = False
        elif ch == '"':
            in_str = True
        elif ch in "{[":
            stack.append("}" if ch == "{" else "]")
        elif ch in "}]" and stack:
            stack.pop()
    if not stack and not in_str:
        return text, False
    fixed = text + ('"' if in_str else "")
    fixed = re.sub(r',\s*"[^"]*"\s*:?\s*$', "", fixed)    # a key with no value yet
    fixed = re.sub(r'[,:]\s*$', "", fixed.rstrip())         # a dangling comma or colon
    return fixed + "".join(reversed(stack)), True


PARSER_STAGES = ("plain", "fences", "extract", "normalize", "truncation")


def parse_json(text: str, stages: tuple[str, ...] = PARSER_STAGES) -> Parsed:
    """Recover one JSON object from model output, applying only the listed stages."""
    repairs: list[str] = []
    candidate = text.strip()
    if "fences" in stages:
        candidate, fenced = strip_fences(candidate)
        if fenced:
            repairs.append("fences")
    if "extract" in stages:
        candidate, cut = extract_object(candidate)
        if cut:
            repairs.append("surrounding_text")
    if "normalize" in stages:
        candidate, fixes = normalize(candidate)
        repairs += fixes
    if "truncation" in stages:
        candidate, closed = close_truncated(candidate)
        if closed:
            repairs.append("closed_truncated")
    try:
        value = json.loads(candidate)
    except json.JSONDecodeError as exc:
        raise ParseError(f"{exc.msg} at char {exc.pos}") from exc
    if not isinstance(value, dict):
        raise ParseError(f"expected an object, got {type(value).__name__}")
    return Parsed(value, repairs)


# ---------------------------------------------------------------------------
# 3. Validation, repair loop with a ceiling, and salvage
# ---------------------------------------------------------------------------


def validate_output(data: dict[str, Any], model_cls: type[BaseModel]) -> BaseModel:
    """Validate with pydantic after turning strict-mode nulls back into defaults."""
    cleaned = {
        k: v for k, v in data.items()
        if not (v is None and k in model_cls.model_fields and not model_cls.model_fields[k].is_required())
    }
    return model_cls.model_validate(cleaned)


def describe_errors(exc: ValidationError) -> list[str]:
    """Short, model-readable error lines such as "priority: Input should be ..."."""
    lines = []
    for err in exc.errors():
        where = ".".join(str(p) for p in err["loc"]) or "(root)"
        lines.append(f"{where}: {err['msg']}")
    return lines


@dataclass
class StructuredResult:
    value: BaseModel | None
    attempts: int
    errors: list[list[str]]            # the errors of each failed attempt
    repairs: list[str]                 # parser repairs on the attempt that succeeded
    usage: Usage
    last_data: dict[str, Any] | None   # the last parsed object, for salvage


def generate_structured(
    llm: Callable[..., ChatResult],
    messages: list[dict[str, Any]],
    model_cls: type[BaseModel],
    *,
    max_attempts: int = 3,
    **chat_kwargs: Any,
) -> StructuredResult:
    """Ask, parse, validate. On failure, show the model its errors and ask again,
    at most `max_attempts` calls in total. Never loops forever, never raises
    for bad output: the caller decides what a failure means."""
    history = list(messages)
    usage = Usage()
    all_errors: list[list[str]] = []
    last_data: dict[str, Any] | None = None
    for attempt in range(1, max_attempts + 1):
        result = llm(history, **chat_kwargs)
        usage.input_tokens += result.usage.input_tokens
        usage.output_tokens += result.usage.output_tokens
        try:
            parsed = parse_json(result.text)
            last_data = parsed.value
            value = validate_output(parsed.value, model_cls)
            return StructuredResult(value, attempt, all_errors, parsed.repairs, usage, last_data)
        except ParseError as exc:
            errors = [f"(root): could not parse JSON: {exc}"]
        except ValidationError as exc:
            errors = describe_errors(exc)
        all_errors.append(errors)
        history += [
            {"role": "assistant", "content": result.text},
            {"role": "user", "content": "Your reply did not match the schema:\n- " + "\n- ".join(errors)
             + "\nReply again with only the corrected JSON object."},
        ]
    return StructuredResult(None, max_attempts, all_errors, [], usage, last_data)


def salvage(data: dict[str, Any], model_cls: type[BaseModel]) -> tuple[dict[str, Any], dict[str, str]]:
    """Split a parsed object into fields that pass validation and fields that do not.

    Returns (valid_fields, problems). A field is valid if the full validation
    reported no error located at it. Unknown keys and missing required fields
    are listed in problems.
    """
    try:
        return validate_output(data, model_cls).model_dump(), {}
    except ValidationError as exc:
        problems: dict[str, str] = {}
        for err in exc.errors():
            name = str(err["loc"][0]) if err["loc"] else "(root)"
            problems.setdefault(name, err["msg"])
        valid = {k: v for k, v in data.items() if k in model_cls.model_fields and k not in problems}
        return valid, problems

"""See the exact JSON that llm.chat sends, without an API key or network.

`capture_wire` swaps the HTTP transport under the openai client for a fake one
that records each request body and answers with a canned response. The request
is real (built by llm.chat and the openai library); only the reply is canned.
"""
from __future__ import annotations

import json
from collections.abc import Iterator
from contextlib import contextmanager
from typing import Any

import httpx
from openai import OpenAI

import supportdesk.llm as llm


@contextmanager
def capture_wire(canned_message: dict[str, Any], finish_reason: str = "stop") -> Iterator[list[dict[str, Any]]]:
    """Record request bodies sent by llm.chat; reply with `canned_message` every time."""
    bodies: list[dict[str, Any]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        bodies.append(json.loads(request.content))
        return httpx.Response(200, json={
            "id": "chatcmpl-canned", "object": "chat.completion", "created": 0, "model": bodies[-1]["model"],
            "choices": [{"index": 0, "message": {"role": "assistant", **canned_message}, "finish_reason": finish_reason}],
            "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
        })

    def fake_client(provider: str) -> OpenAI:
        return OpenAI(api_key="not-a-real-key", base_url=llm.PROVIDERS[provider]["base_url"],
                      http_client=httpx.Client(transport=httpx.MockTransport(handler)))

    original = llm.make_client
    llm.make_client = fake_client
    try:
        yield bodies
    finally:
        llm.make_client = original

"""The request body llm.chat sends for native structured output (captured offline)."""
from __future__ import annotations

import json

from examples.m06_native import native_messages
from examples.m06_structured import response_format_for
from examples.m06_wire import capture_wire
from supportdesk.llm import chat
from supportdesk.schemas import Triage

canned = {"content": '{"category":"billing","priority":"high","language":"en",'
                     '"summary":"Customer was charged twice for the Team plan.","needs_human":true}'}
with capture_wire(canned) as bodies:
    result = chat(native_messages("Subject: Charged twice\n\nMy card was charged 288 USD twice."),
                  provider="groq", response_format=response_format_for(Triage), max_tokens=2000)

body = bodies[0]
print(json.dumps({k: v for k, v in body.items() if k != "response_format"}, indent=2))
print("response_format.type:", body["response_format"]["type"])
print("response_format.json_schema keys:", list(body["response_format"]["json_schema"]))
print("\nChatResult.text:", result.text)
print("finish_reason:", result.finish_reason, "| provider:", result.provider, "| model:", result.model)

{
  "messages": [
    {
      "role": "system",
      "content": "You triage Brightlane support tickets. Brightlane is a project-management SaaS."
    },
    {
      "role": "user",
      "content": "Subject: Charged twice\n\nMy card was charged 288 USD twice."
    }
  ],
  "model": "openai/gpt-oss-120b",
  "max_tokens": 2000,
  "temperature": 0.0
}
response_format.type: json_schema
response_format.json_schema keys: ['name', 'schema', 'strict']

ChatResult.text: {"category":"billing","priority":"high","language":"en","summary":"Customer was charged twice for the Team plan.","needs_human":true}
finish_reason: stop | provider: groq | model: openai/gpt-oss-120b

"""Native structured output vs prompted JSON for ticket triage, through llm.chat.

Needs a provider key (or a local Ollama). Usage:
  LLM_PROVIDER=groq GROQ_API_KEY=... python examples/m06_native.py
"""
from __future__ import annotations

import json
import os
import sys

from pydantic import ValidationError

from examples.m06_structured import ParseError, parse_json, response_format_for, validate_output
from supportdesk.data import load_tickets
from supportdesk.llm import PROVIDERS, chat, resolve
from supportdesk.schemas import Triage, triage_json_schema

SYSTEM = "You triage Brightlane support tickets. Brightlane is a project-management SaaS."


def native_messages(ticket_text: str) -> list[dict]:
    return [{"role": "system", "content": SYSTEM}, {"role": "user", "content": ticket_text}]


def prompted_messages(ticket_text: str) -> list[dict]:
    schema = json.dumps(triage_json_schema(), separators=(",", ":"))
    return [{"role": "system", "content": f"{SYSTEM}\nReply with only a JSON object that matches this JSON Schema, "
             f"with no other text:\n{schema}"}, {"role": "user", "content": ticket_text}]


def triage_once(ticket_text: str, mode: str) -> tuple[str, str]:
    """Return (outcome, detail) for one ticket in 'native' or 'prompted' mode."""
    if mode == "native":
        result = chat(native_messages(ticket_text), response_format=response_format_for(Triage), max_tokens=2000)
    else:
        result = chat(prompted_messages(ticket_text), max_tokens=2000)
    try:
        parsed = parse_json(result.text)
        triage = validate_output(parsed.value, Triage)
        return "valid", f"{triage.category}/{triage.priority} repairs={parsed.repairs} out_tokens={result.usage.output_tokens}"
    except (ParseError, ValidationError) as exc:
        return "invalid", str(exc).splitlines()[0]


if __name__ == "__main__":
    provider, model = resolve()
    key_env = PROVIDERS[provider]["key_env"]
    if key_env and not os.environ.get(key_env):
        sys.exit(f"Set {key_env} (or LLM_PROVIDER=ollama) to run this example against {provider}/{model}.")
    print(f"provider={provider} model={model}")
    for ticket in load_tickets("dev")[:5]:
        for mode in ("native", "prompted"):
            outcome, detail = triage_once(ticket.text, mode)
            print(f"{ticket.id} {mode:<8} {outcome:<7} {detail}  gold={ticket.gold['category']}/{ticket.gold['priority']}")

Set GROQ_API_KEY (or LLM_PROVIDER=ollama) to run this example against groq/openai/gpt-oss-120b.

provider=groq model=openai/gpt-oss-120b
T-1001 native   valid   billing/high repairs=[] out_tokens=...  gold=billing/high
T-1001 prompted valid   billing/high repairs=[] out_tokens=...  gold=billing/high
T-1002 native   valid   billing/normal repairs=[] out_tokens=...  gold=billing/normal

"""A corpus of 24 hostile model outputs for a Triage object, and the parser scoreboard.

Each case imitates a failure seen in real model output. Run it to see how many
cases each parser stage recovers, and how many then pass Triage validation.
"""
from __future__ import annotations

from pydantic import ValidationError

from examples.m06_structured import PARSER_STAGES, ParseError, parse_json, validate_output
from supportdesk.schemas import Triage

GOOD = '{"category": "billing", "priority": "high", "language": "en", "summary": "Customer was charged twice for the Team plan.", "needs_human": true}'

HOSTILE: dict[str, str] = {
    "clean": GOOD,
    "clean_pretty": '{\n  "category": "how_to",\n  "priority": "normal",\n  "language": "de",\n  "summary": "Asks how to export a board to CSV.",\n  "needs_human": false\n}',
    "fence_json": "```json\n" + GOOD + "\n```",
    "fence_bare": "```\n" + GOOD + "\n```",
    "preamble": "Sure! Here is the triage for this ticket:\n\n" + GOOD,
    "postamble": GOOD + "\n\nLet me know if you'd like me to draft a reply as well.",
    "preamble_fence_post": "Here's the JSON you asked for:\n```json\n" + GOOD + "\n```\nI marked it high because money is wrong.",
    "trailing_comma": '{"category": "billing", "priority": "high", "language": "en", "summary": "Customer was charged twice.", "needs_human": true,}',
    "trailing_comma_pretty": '{\n  "category": "bug",\n  "priority": "urgent",\n  "language": "en",\n  "summary": "Boards fail to load for the whole team.",\n  "needs_human": true,\n}',
    "single_quotes": "{'category': 'billing', 'priority': 'high', 'language': 'en', 'summary': 'Customer was charged twice.', 'needs_human': true}",
    "python_dict": "{'category': 'bug', 'priority': 'high', 'language': 'es', 'summary': 'Mobile app crashes on login.', 'needs_human': False}",
    "smart_quotes": "{\u201ccategory\u201d: \u201cbilling\u201d, \u201cpriority\u201d: \u201chigh\u201d, \u201clanguage\u201d: \u201cen\u201d, \u201csummary\u201d: \u201cCustomer was charged twice.\u201d, \u201cneeds_human\u201d: true}",
    "braces_in_string": 'Result: {"category": "how_to", "priority": "low", "language": "en", "summary": "Asks what {{name}} means in templates.", "needs_human": false} Done.',
    "truncated_in_value": '{"category": "billing", "priority": "high", "language": "en", "summary": "Customer was charged tw',
    "truncated_after_comma": '{"category": "billing", "priority": "high", "language": "en", "summary": "Customer was charged twice.",',
    "truncated_in_key": '{"category": "account_access", "priority": "high", "language": "ja", "summary": "Cannot log in after SSO change.", "needs_hu',
    "fence_truncated": "```json\n{\"category\": \"cancellation\", \"priority\": \"normal\", \"language\": \"en\", \"summary\": \"Wants to cancel the annual plan",
    "wrong_enum": '{"category": "Billing", "priority": "critical", "language": "en", "summary": "Customer was charged twice.", "needs_human": true}',
    "wrong_types": '{"category": "billing", "priority": "high", "language": "English", "summary": "Charged twice.", "needs_human": "yes"}',
    "extra_field": '{"category": "billing", "priority": "high", "language": "en", "summary": "Customer was charged twice.", "needs_human": true, "confidence": 0.9}',
    "array_wrapped": '[' + GOOD + ']',
    "prose_only": "This looks like a billing problem with high priority; a human should refund the duplicate charge.",
    "refusal": "I'm sorry, but I can't help with that request.",
    "apostrophe_in_single": "{'category': 'billing', 'priority': 'high', 'language': 'en', 'summary': 'Customer's card was charged twice.', 'needs_human': true}",
}

LADDER = [PARSER_STAGES[:i] for i in range(1, len(PARSER_STAGES) + 1)]


def score(stages: tuple[str, ...]) -> tuple[int, int, list[str]]:
    """How many cases parse to an object, and how many of those validate as Triage."""
    parsed = valid = 0
    failures = []
    for name, text in HOSTILE.items():
        try:
            data = parse_json(text, stages).value
        except ParseError:
            failures.append(name)
            continue
        parsed += 1
        try:
            validate_output(data, Triage)
            valid += 1
        except ValidationError:
            pass
    return parsed, valid, failures


if __name__ == "__main__":
    n = len(HOSTILE)
    print(f"{n} hostile outputs\n")
    print(f"{'parser stages':<50} {'parsed':>10} {'valid Triage':>14}")
    for stages in LADDER:
        parsed, valid, _ = score(stages)
        print(f"{' + '.join(stages):<50} {parsed:>3}/{n} {parsed / n:>4.0%} {valid:>6}/{n} {valid / n:>4.0%}")
    print("\nstill unparseable with every stage:", score(PARSER_STAGES)[2])
    print("\nrepairs logged per case (full parser):")
    for name, text in HOSTILE.items():
        try:
            print(f"  {name:<22} {parse_json(text).repairs}")
        except ParseError as exc:
            print(f"  {name:<22} ParseError: {exc}")

24 hostile outputs

parser stages                                          parsed   valid Triage
plain                                                5/24  21%      2/24   8%
plain + fences                                       8/24  33%      5/24  21%
plain + fences + extract                            12/24  50%      9/24  38%
plain + fences + extract + normalize                17/24  71%     14/24  58%
plain + fences + extract + normalize + truncation   21/24  88%     14/24  58%

still unparseable with every stage: ['prose_only', 'refusal', 'apostrophe_in_single']

repairs logged per case (full parser):
  clean                  []
  clean_pretty           []
  fence_json             ['fences']
  fence_bare             ['fences']
  preamble               ['surrounding_text']
  postamble              ['surrounding_text']
  preamble_fence_post    ['fences']
  trailing_comma         ['trailing_comma']
  trailing_comma_pretty  ['trailing_comma']
  single_quotes          ['single_quotes']
  python_dict            ['single_quotes', 'python_literals']
  smart_quotes           ['smart_quotes']
  braces_in_string       ['surrounding_text']
  truncated_in_value     ['closed_truncated']
  truncated_after_comma  ['closed_truncated']
  truncated_in_key       ['closed_truncated']
  fence_truncated        ['fences', 'closed_truncated']
  wrong_enum             []
  wrong_types            []
  extra_field            []
  array_wrapped          ['surrounding_text']
  prose_only             ParseError: no '{' in output
  refusal                ParseError: no '{' in output
  apostrophe_in_single   ParseError: Expecting ',' delimiter at char 83

"""The repair loop with an attempt ceiling, driven by ScriptedLLM (plumbing only, not a model)."""
from __future__ import annotations

from examples.m06_structured import generate_structured
from supportdesk.schemas import Triage
from supportdesk.stand_in import ScriptedLLM
from supportdesk.tokens import count_messages

messages = [
    {"role": "system", "content": "Triage the ticket. Reply with only a JSON object for the Triage schema."},
    {"role": "user", "content": "Subject: Charged twice\n\nMy card was charged 288 USD twice on 3 September."},
]
BROKEN = ('Here you go: {"category": "billing", "priority": "critical", "language": "English", '
          '"summary": "Charged twice for the Team plan."}')
FIXED = ('{"category": "billing", "priority": "high", "language": "en", '
         '"summary": "Charged twice for the Team plan.", "needs_human": true}')

print("=== Case 1: broken, then fixed")
llm = ScriptedLLM(replies=[BROKEN, FIXED])
result = generate_structured(llm, messages, Triage, max_attempts=3)
print("value:", result.value)
print("attempts:", result.attempts, "| errors per failed attempt:", result.errors)
print("tokens spent:", result.usage.input_tokens, "in,", result.usage.output_tokens, "out")
print("the correction message the loop sent on attempt 2:")
print("  " + llm.calls[1]["messages"][-1]["content"].replace("\n", "\n  "))

print("\n=== Case 2: never fixed, the ceiling stops it")
stubborn = ScriptedLLM(responder=lambda msgs, kw: BROKEN)
result = generate_structured(stubborn, messages, Triage, max_attempts=3)
print("value:", result.value, "| attempts:", result.attempts, "| calls made:", len(stubborn.calls))
print("input tokens per call (estimate):", [count_messages(c["messages"]) for c in stubborn.calls])
print("tokens spent:", result.usage.input_tokens, "in,", result.usage.output_tokens, "out")
print("last parsed object kept for salvage:", result.last_data)

=== Case 1: broken, then fixed
value: category='billing' priority='high' language='en' summary='Charged twice for the Team plan.' needs_human=True
attempts: 2 | errors per failed attempt: [["priority: Input should be 'low', 'normal', 'high' or 'urgent'", "language: String should match pattern '^[a-z]{2}$'", 'needs_human: Field required']]
tokens spent: 196 in, 72 out
the correction message the loop sent on attempt 2:
  Your reply did not match the schema:
  - priority: Input should be 'low', 'normal', 'high' or 'urgent'
  - language: String should match pattern '^[a-z]{2}$'
  - needs_human: Field required
  Reply again with only the corrected JSON object.

=== Case 2: never fixed, the ceiling stops it
value: None | attempts: 3 | calls made: 3
input tokens per call (estimate): [47, 149, 251]
tokens spent: 447 in, 105 out
last parsed object kept for salvage: {'category': 'billing', 'priority': 'critical', 'language': 'English', 'summary': 'Charged twice for the Team plan.'}

"""Salvage: keep the valid fields of an invalid object, then apply a safe fallback policy."""
from __future__ import annotations

from examples.m06_hostile import HOSTILE
from examples.m06_structured import ParseError, parse_json, salvage
from supportdesk.schemas import Triage

SAFE_DEFAULTS = {"priority": "high", "needs_human": True}   # unknown means: a person looks, soon


def triage_or_fallback(data: dict) -> tuple[str, dict, dict]:
    """Return (status, triage_fields, problems). status is valid, salvaged, or unusable."""
    valid, problems = salvage(data, Triage)
    if not problems:
        return "valid", valid, {}
    if "category" not in valid:                      # without a category we cannot route at all
        return "unusable", valid, problems
    filled = {**valid, **{k: v for k, v in SAFE_DEFAULTS.items() if k not in valid}}
    filled["needs_human"] = True                     # anything salvaged gets human review
    return "salvaged", filled, problems


counts = {"valid": 0, "salvaged": 0, "unusable": 0, "unparseable": 0}
for name, text in HOSTILE.items():
    try:
        parsed = parse_json(text)
    except ParseError:
        counts["unparseable"] += 1
        continue
    status, fields, problems = triage_or_fallback(parsed.value)
    if "closed_truncated" in parsed.repairs and status == "valid":
        status = "salvaged"                          # a closed truncation is never fully trusted
    counts[status] += 1
    if status != "valid":
        route = {k: fields.get(k) for k in ("category", "priority", "needs_human")}
        print(f"{name:<22} {status:<9} route with {route}")
        print(f"{'':<32} problems={problems}")
print("\n", counts, f"(n={len(HOSTILE)})")

truncated_in_value     salvaged  route with {'category': 'billing', 'priority': 'high', 'needs_human': True}
                                 problems={'needs_human': 'Field required'}
truncated_after_comma  salvaged  route with {'category': 'billing', 'priority': 'high', 'needs_human': True}
                                 problems={'needs_human': 'Field required'}
truncated_in_key       salvaged  route with {'category': 'account_access', 'priority': 'high', 'needs_human': True}
                                 problems={'needs_human': 'Field required'}
fence_truncated        salvaged  route with {'category': 'cancellation', 'priority': 'normal', 'needs_human': True}
                                 problems={'needs_human': 'Field required'}
wrong_enum             unusable  route with {'category': None, 'priority': None, 'needs_human': True}
                                 problems={'category': "Input should be 'billing', 'cancellation', 'account_access', 'bug', 'how_to' or 'feature_request'", 'priority': "Input should be 'low', 'normal', 'high' or 'urgent'"}
wrong_types            salvaged  route with {'category': 'billing', 'priority': 'high', 'needs_human': True}
                                 problems={'language': "String should match pattern '^[a-z]{2}$'"}
extra_field            salvaged  route with {'category': 'billing', 'priority': 'high', 'needs_human': True}
                                 problems={'confidence': 'Extra inputs are not permitted'}

 {'valid': 14, 'salvaged': 6, 'unusable': 1, 'unparseable': 3} (n=24)

"""What a schema costs in tokens: formatting, descriptions, width, and nesting depth."""
from __future__ import annotations

import json
from typing import Any

from pydantic import BaseModel, ConfigDict, create_model

from examples.m06_structured import strict_schema
from supportdesk.schemas import Triage
from supportdesk.tokens import count_tokens


def tokens(schema: dict[str, Any], indent: int | None = None) -> int:
    text = json.dumps(schema, indent=indent, separators=None if indent else (",", ":"))
    return count_tokens(text, "o200k_base")


def without(schema: Any, keys: set[str]) -> Any:
    if isinstance(schema, dict):
        return {k: without(v, keys) for k, v in schema.items() if k not in keys}
    if isinstance(schema, list):
        return [without(v, keys) for v in schema]
    return schema


base = Triage.model_json_schema()
print("Triage schema, o200k_base tokens")
print(f"  pydantic output, indent=2          {tokens(base, 2):>5}")
print(f"  compact separators                 {tokens(base):>5}")
print(f"  strict (titles removed)            {tokens(strict_schema(Triage)):>5}")
print(f"  strict, descriptions removed       {tokens(without(strict_schema(Triage), {'description'})):>5}")

# Width: the same kind of field, repeated.
print("\nWidth: flat objects with N string fields (each with a 10-word description)")
for n in (5, 10, 20, 40):
    fields = {f"field_{i}": (str, ...) for i in range(n)}
    model = create_model(f"Wide{n}", __config__=ConfigDict(extra="forbid"), **fields)
    schema = strict_schema(model)
    for prop in schema["properties"].values():
        prop["description"] = "One short sentence the support agent will read before replying."
    print(f"  {n:>2} fields: {tokens(schema):>5} tokens")


# Depth: the same five Triage leaves, wrapped in 0 to 3 extra levels of objects.
def nest(model: type[BaseModel], depth: int) -> type[BaseModel]:
    for level in range(depth):
        model = create_model(f"Level{level}", __config__=ConfigDict(extra="forbid"), inner=(model, ...))
    return model


print("\nDepth: Triage's five fields wrapped in extra object levels")
for depth in range(4):
    schema = strict_schema(nest(Triage, depth))
    example = json.loads(Triage(category="billing", priority="high", language="en",
                                summary="Customer was charged twice.", needs_human=True).model_dump_json())
    for _ in range(depth):
        example = {"inner": example}
    print(f"  depth {depth + 1}: schema {tokens(schema):>4} tokens (uses $defs: {'$defs' in schema}), "
          f"one answer {count_tokens(json.dumps(example), 'o200k_base'):>3} tokens")

Triage schema, o200k_base tokens
  pydantic output, indent=2            381
  compact separators                   248
  strict (titles removed)              222
  strict, descriptions removed         113

Width: flat objects with N string fields (each with a 10-word description)
   5 fields:   146 tokens
  10 fields:   276 tokens
  20 fields:   536 tokens
  40 fields:  1056 tokens

Depth: Triage's five fields wrapped in extra object levels
  depth 1: schema  222 tokens (uses $defs: False), one answer  34 tokens
  depth 2: schema  257 tokens (uses $defs: True), one answer  37 tokens
  depth 3: schema  290 tokens (uses $defs: True), one answer  41 tokens
  depth 4: schema  323 tokens (uses $defs: True), one answer  44 tokens

"""Does a nested schema change accuracy? A harness to measure it on the dev tickets.

  LLM_PROVIDER=groq GROQ_API_KEY=... PYTHONPATH=. python examples/m06_schema_eval.py
  PYTHONPATH=. python examples/m06_schema_eval.py --dry     (keyword baseline: checks the harness only)
"""
from __future__ import annotations

import json
import sys

from pydantic import BaseModel, ConfigDict

from examples.m06_lab import keyword_triage, wilson
from examples.m06_structured import generate_structured, response_format_for
from supportdesk.data import load_tickets
from supportdesk.schemas import Category, Priority, Triage
from supportdesk.stand_in import ScriptedLLM


class Routing(BaseModel):
    model_config = ConfigDict(extra="forbid")
    category: Category
    priority: Priority
    needs_human: bool


class Customer(BaseModel):
    model_config = ConfigDict(extra="forbid")
    language: str
    summary: str


class NestedTriage(BaseModel):
    """The same five facts as Triage, grouped two levels deep."""
    model_config = ConfigDict(extra="forbid")
    routing: Routing
    customer: Customer


def category_of(value: BaseModel) -> str:
    return value.routing.category if isinstance(value, NestedTriage) else value.category


def run(schema_cls: type[BaseModel], llm, tickets) -> tuple[int, int, int]:
    valid = correct = out_tokens = 0
    for t in tickets:
        messages = [{"role": "system", "content": "Triage this Brightlane support ticket as JSON."},
                    {"role": "user", "content": t.text}]
        res = generate_structured(llm, messages, schema_cls, max_attempts=1,
                                  response_format=response_format_for(schema_cls), max_tokens=2000)
        out_tokens += res.usage.output_tokens
        if res.value is not None:
            valid += 1
            correct += category_of(res.value) == t.gold["category"]
    return valid, correct, out_tokens


if __name__ == "__main__":
    tickets = load_tickets("dev")
    if "--dry" in sys.argv:
        by_text = {t.text: t for t in tickets}

        def dry(messages, kwargs):
            d = keyword_triage(by_text[messages[1]["content"]])
            if kwargs["response_format"]["json_schema"]["name"] == "NestedTriage":
                d = {"routing": {k: d[k] for k in ("category", "priority", "needs_human")},
                     "customer": {k: d[k] for k in ("language", "summary")}}
            return json.dumps(d)
        llm = ScriptedLLM(responder=dry)
    else:
        from supportdesk.llm import chat as llm
    n = len(tickets)
    for schema_cls in (Triage, NestedTriage):
        valid, correct, out_tokens = run(schema_cls, llm, tickets)
        lo, hi = wilson(correct, n)
        print(f"{schema_cls.__name__:<13} valid {valid}/{n}  category accuracy {correct}/{n} = {correct / n:.0%} "
              f"(95% CI {lo:.0%} to {hi:.0%})  output tokens {out_tokens}")