cp -r supportdesk work/m14 && cd work/m14
source /home/claude/venv/bin/activate          # or your own Python 3.11 venv
pip install scikit-learn==1.9.1
PYTHONPATH=. python -c "import sklearn; print(sklearn.__version__)"

# examples/m14_common.py
"""Shared setup for the Module 14 examples (run every example from the repo root with PYTHONPATH=.).

Every example works without an API key: it uses a ScriptedLLM stand-in by default.
Set M14_LIVE=1 (plus LLM_PROVIDER and the provider's key) to send the same
requests to a real model through supportdesk.llm.chat instead.
"""
from __future__ import annotations

import math
import os
from collections.abc import Callable

from supportdesk.llm import chat, resolve
from supportdesk.stand_in import ScriptedLLM

ChatFn = Callable[..., object]


def pick_llm(stand_in: ScriptedLLM) -> ChatFn:
    """Return the real `chat` when M14_LIVE=1, otherwise the scripted stand-in."""
    if os.environ.get("M14_LIVE") == "1":
        provider, model = resolve()
        print(f"[live] using {provider}:{model}")
        return chat
    print("[stand-in] ScriptedLLM: tests the plumbing only, not model quality")
    return stand_in


def wilson(k: int, n: int, z: float = 1.96) -> tuple[float, float]:
    """95% Wilson confidence interval for a proportion k/n (honest with small n)."""
    if n == 0:
        return (0.0, 1.0)
    p = k / n
    centre = (p + z * z / (2 * n)) / (1 + z * z / n)
    half = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / (1 + z * z / n)
    return (max(0.0, centre - half), min(1.0, centre + half))


def fmt_rate(k: int, n: int) -> str:
    """'17/24 = 70.8% (95% CI 50.8% to 85.1%)'."""
    lo, hi = wilson(k, n)
    return f"{k}/{n} = {k / n:.1%} (95% CI {lo:.1%} to {hi:.1%})"

# examples/m14_triage_at_scale.py
"""Pattern 1: classification and extraction at scale (Brightlane ticket triage).

Runs 72 tickets through a triage prompt with a thread pool, validates every reply
against the Triage schema, sends failures to a dead-letter list, and projects the
monthly bill for 100,000 tickets with supportdesk.pricing.
"""
from __future__ import annotations

import json
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass

from pydantic import ValidationError

from examples.m14_common import pick_llm
from supportdesk.data import CATEGORIES, Ticket, load_tickets
from supportdesk.llm import ChatResult, Usage
from supportdesk.pricing import cost_usd
from supportdesk.schemas import Triage
from supportdesk.stand_in import ScriptedLLM
from supportdesk.tokens import count_messages, count_tokens

SYSTEM = (
    "You triage support tickets for Brightlane, a project-management SaaS.\n"
    f"Categories: {', '.join(CATEGORIES)}.\n"
    "Priorities: urgent (many users blocked or security risk now), high (one user blocked or money wrong), "
    "normal (needs an answer), low (question or idea).\n"
    "Reply with only a JSON object with keys category, priority, language (ISO 639-1), "
    "summary (one English sentence), needs_human (true or false)."
)


def build_messages(ticket: Ticket) -> list[dict]:
    return [{"role": "system", "content": SYSTEM},
            {"role": "user", "content": f"<ticket>\n{ticket.text}\n</ticket>"}]


@dataclass
class Outcome:
    ticket_id: str
    triage: Triage | None
    error: str
    usage: Usage
    latency_ms: float


def triage_one(ticket: Ticket, llm) -> Outcome:
    started = time.perf_counter()
    result: ChatResult = llm(build_messages(ticket), temperature=0.0, max_tokens=300)
    ms = (time.perf_counter() - started) * 1000
    try:
        return Outcome(ticket.id, Triage.model_validate_json(result.text), "", result.usage, ms)
    except ValidationError as exc:
        return Outcome(ticket.id, None, exc.errors()[0]["type"], result.usage, ms)


def triage_many(tickets: list[Ticket], llm, workers: int) -> tuple[list[Outcome], float]:
    started = time.perf_counter()
    with ThreadPoolExecutor(max_workers=workers) as pool:
        outcomes = list(pool.map(lambda t: triage_one(t, llm), tickets))
    return outcomes, time.perf_counter() - started


def scripted_responder(messages, kwargs):
    """STAND-IN: echoes gold labels after a fake 50 ms 'network' delay; every 12th reply is broken."""
    time.sleep(0.05)
    text = messages[-1]["content"]
    ticket = next(t for t in TICKETS if t.text in text)
    if int(ticket.id[-2:]) % 12 == 0:
        return '{"category": "billing", "priority": "normal"'  # truncated JSON, like a max_tokens cut
    return json.dumps({"category": ticket.gold["category"], "priority": ticket.gold["priority"],
                       "language": ticket.language, "summary": f"Customer asks about: {ticket.subject}.",
                       "needs_human": not ticket.gold["answerable"]})


TICKETS = load_tickets()

if __name__ == "__main__":
    llm = pick_llm(ScriptedLLM(responder=scripted_responder))
    count_tokens("warm up the tokenizer so the first timing is fair")

    for workers in (1, 8):
        outcomes, seconds = triage_many(TICKETS, llm, workers)
        dead = [o for o in outcomes if o.triage is None]
        print(f"workers={workers}: {len(TICKETS)} tickets in {seconds:.2f} s, "
              f"valid {len(outcomes) - len(dead)}, dead-letter {len(dead)} {[(o.ticket_id, o.error) for o in dead]}")

    # Cost projection from measured prompt sizes (o200k_base token counts, an estimate for other tokenizers).
    input_tokens = [count_messages(build_messages(t)) for t in TICKETS]
    system_tokens = count_tokens(SYSTEM) + 4
    example_reply = '{"category": "billing", "priority": "high", "language": "en", "summary": "Customer was charged twice for the Team plan and wants the duplicate refunded.", "needs_human": true}'
    out_tokens = count_tokens(example_reply)
    avg_in = sum(input_tokens) / len(input_tokens)
    print(f"\nmeasured: avg input {avg_in:.0f} tokens (system prefix {system_tokens}), reply {out_tokens} tokens")

    monthly = 100_000
    reasoning = {"openai/gpt-oss-120b": 150, "openai/gpt-oss-20b": 150, "gemini-3.5-flash": 150,
                 "gemini-3.5-flash-lite": 0, "llama-3.1-8b-instant": 0}  # ASSUMED hidden reasoning tokens per call
    print(f"{'model':24} {'per call':>10} {'100k/month':>11} {'batch API':>10} {'cached prefix':>13}")
    for model, extra in reasoning.items():
        usage = Usage(input_tokens=round(avg_in), output_tokens=out_tokens + extra)
        cached = Usage(input_tokens=round(avg_in), cached_tokens=system_tokens, output_tokens=out_tokens + extra)
        one = cost_usd(usage, model)
        print(f"{model:24} {one:>10.6f} {one * monthly:>11.2f} {cost_usd(usage, model, batch=True) * monthly:>10.2f} "
              f"{cost_usd(cached, model) * monthly:>13.2f}")

    # Packing 10 tickets into one request amortizes the system prompt.
    packed_in = system_tokens + 3 + sum(input_tokens[:10]) - 10 * (system_tokens + 3)
    print(f"\npacking 10 tickets per request: {sum(input_tokens[:10])} input tokens as 10 calls "
          f"vs about {packed_in} as one call ({1 - packed_in / sum(input_tokens[:10]):.0%} fewer)")

# examples/m14_weekly_digest.py
"""Pattern 2: summarization and document processing (Maya's weekly ticket digest).

Code computes every number; the model only writes prose around them (map per
category, then reduce). A checker then refuses any digest whose numbers are not
in the computed facts.
"""
from __future__ import annotations

import json
import re
from collections import Counter

from examples.m14_common import pick_llm
from supportdesk.data import Ticket, load_tickets
from supportdesk.stand_in import ScriptedLLM


def weekly_facts(tickets: list[Ticket]) -> dict:
    """Deterministic statistics: the part of a digest that must never be wrong."""
    return {
        "total": len(tickets),
        "by_category": dict(Counter(t.gold["category"] for t in tickets).most_common()),
        "urgent_or_high": sum(t.gold["priority"] in ("urgent", "high") for t in tickets),
        "non_english": sum(t.language != "en" for t in tickets),
        "not_answerable_from_kb": sum(not t.gold["answerable"] for t in tickets),
    }


def map_prompt(category: str, tickets: list[Ticket]) -> list[dict]:
    lines = "\n".join(f"- [{t.id}] {t.subject}: {t.body[:160]}" for t in tickets)
    return [{"role": "system", "content": "Summarize the common themes in these support tickets in at most 2 "
                                          "sentences. Mention ticket ids for examples. Do not state counts."},
            {"role": "user", "content": f"Category: {category}\n{lines}"}]


def reduce_prompt(facts: dict, themes: dict[str, str]) -> list[dict]:
    return [{"role": "system", "content": "Write a weekly support digest for the support lead in under 150 words. "
                                          "Use ONLY numbers that appear in FACTS. Do not compute new numbers."},
            {"role": "user", "content": f"FACTS: {json.dumps(facts)}\nTHEMES: {json.dumps(themes)}"}]


def allowed_numbers(facts: dict) -> set[str]:
    return set(re.findall(r"\d+", json.dumps(facts)))


def check_numbers(digest: str, facts: dict) -> list[str]:
    """Every number in the digest must appear in the facts (ticket ids like T-1040 are allowed)."""
    text = re.sub(r"T-\d{4}", "", digest)
    return [n for n in re.findall(r"\d+", text) if n not in allowed_numbers(facts)]


# STAND-IN replies (written by hand to exercise the pipeline; not model output).
MAP_REPLY = "Themes vary; see for example [{first}] and [{second}]."
REDUCE_REPLY = (
    "This week we received 72 tickets. How-to questions led (18), followed by account access (17) and "
    "billing (16). 17 tickets were urgent or high priority, and 8 arrived in a language other than English. "
    "Refund and duplicate-charge requests (T-1001, T-1031) remain the most time-sensitive billing theme. "
    "11 tickets could not be answered from the help center, which points to article gaps."
)


def responder(messages, kwargs):
    if "Category:" in messages[-1]["content"]:
        ids = re.findall(r"\[(T-\d{4})\]", messages[-1]["content"])
        return MAP_REPLY.format(first=ids[0], second=ids[-1])
    return REDUCE_REPLY


if __name__ == "__main__":
    tickets = load_tickets()
    facts = weekly_facts(tickets)
    print("FACTS", json.dumps(facts))
    llm = pick_llm(ScriptedLLM(responder=responder))

    themes = {}
    for category in facts["by_category"]:
        group = [t for t in tickets if t.gold["category"] == category]
        themes[category] = llm(map_prompt(category, group), temperature=0.2).text
    digest = llm(reduce_prompt(facts, themes), temperature=0.2).text
    print(f"\nmap calls: {len(themes)}, reduce calls: 1")
    print("\nDIGEST\n" + digest)
    bad = check_numbers(digest, facts)
    print("\nnumber check:", "PASS" if not bad else f"FAIL, numbers not in facts: {bad}")

# examples/m14_grounded_answer.py
"""Pattern 3: a conversational assistant grounded in the help center, with citations and abstention.

Two gates protect the customer: a retrieval gate (abstain when the best BM25
score is weak, threshold chosen on dev and measured on test) and a citation
gate (every cited article must be one we actually retrieved and showed).
"""
from __future__ import annotations

from pydantic import ValidationError

from examples.m14_common import fmt_rate, pick_llm
from supportdesk.data import Ticket, get_article, load_tickets
from supportdesk.kb_search import Hit, KBSearch
from supportdesk.schemas import DraftReply
from supportdesk.stand_in import ScriptedLLM

KB = KBSearch()
ABSTAIN = DraftReply(reply="I could not find this in our help center, so I have passed your question to a "
                           "teammate who will reply personally.", cited_articles=[], confidence="low")


def safe_to_answer(ticket: Ticket, hits: list[Hit]) -> bool:
    """Gold definition of 'answering is safe': answerable AND our top hit is the right article."""
    return bool(hits) and ticket.gold["answerable"] and hits[0].article_id == ticket.gold["kb_article"]


def choose_threshold(tickets: list[Ticket]) -> float:
    """Lowest threshold on dev at which at most 1 in 10 answered tickets would be unsafe."""
    scored = [(KB.search(t.text, 1), t) for t in tickets]
    for threshold in [x / 2 for x in range(0, 41)]:
        answered = [(h, t) for h, t in scored if h and h[0].score >= threshold]
        unsafe = sum(not safe_to_answer(t, h) for h, t in answered)
        if answered and unsafe / len(answered) <= 0.10:
            return threshold
    return float("inf")


def build_messages(ticket: Ticket, hits: list[Hit]) -> list[dict]:
    sources = "\n\n".join(f'<source id="{h.article_id}">\n{get_article(h.article_id).body}\n</source>' for h in hits)
    return [{"role": "system", "content": (
                "Answer the customer using ONLY the sources. Cite the ids you used in cited_articles. "
                "If the sources do not answer the question, say so and set confidence to low. Reply as JSON "
                "with keys reply, cited_articles, confidence (low, medium, high).")},
            {"role": "user", "content": f"{sources}\n\n<ticket>\n{ticket.text}\n</ticket>"}]


def check_citations(draft: DraftReply, hits: list[Hit]) -> list[str]:
    problems = []
    shown = {h.article_id for h in hits}
    if not draft.cited_articles and draft.confidence != "low":
        problems.append("confident answer without citations")
    problems += [f"cites {c} which was not retrieved" for c in draft.cited_articles if c not in shown]
    return problems


def answer(ticket: Ticket, llm, threshold: float) -> tuple[DraftReply, str]:
    hits = KB.search(ticket.text, 3)
    if not hits or hits[0].score < threshold:
        return ABSTAIN, f"abstain: top score {hits[0].score if hits else 0} < {threshold}"
    result = llm(build_messages(ticket, hits), temperature=0.0, response_format={"type": "json_object"})
    try:
        draft = DraftReply.model_validate_json(result.text)
    except ValidationError:
        return ABSTAIN, "abstain: reply failed schema validation"
    problems = check_citations(draft, hits)
    if problems:
        return ABSTAIN, f"abstain: {problems}"
    return draft, "answered"


# STAND-IN replies (hand-written, not model output): one good, one citing an article it was never shown.
SCRIPTED = [
    '{"reply": "Invoices can be reissued with a new address: update Settings > Billing > Tax details, then ask us '
    'to reissue past invoices.", "cited_articles": ["billing-invoices"], "confidence": "high"}',
    '{"reply": "Yes, Team includes SSO.", "cited_articles": ["billing-plans-2025"], "confidence": "high"}',
]

if __name__ == "__main__":
    dev, test = load_tickets("dev"), load_tickets("test")
    threshold = choose_threshold(dev)
    print(f"threshold chosen on dev: {threshold}")
    for name, split in (("dev", dev), ("test", test)):
        answered = [t for t in split if KB.search(t.text, 1) and KB.search(t.text, 1)[0].score >= threshold]
        unsafe = sum(not safe_to_answer(t, KB.search(t.text, 1)) for t in answered)
        missed = sum(safe_to_answer(t, KB.search(t.text, 1)) for t in split if t not in answered)
        print(f"{name}: answered {fmt_rate(len(answered), len(split))}; unsafe among answered "
              f"{fmt_rate(unsafe, len(answered))}; safe tickets we abstained on: {missed}")

    llm = pick_llm(ScriptedLLM(replies=list(SCRIPTED)))
    for ticket_id in ("T-1041", "T-1012", "T-1036"):
        ticket = next(t for t in dev + test if t.id == ticket_id)
        draft, why = answer(ticket, llm, threshold)
        print(f"\n{ticket_id} ({why})\n  {draft.reply}\n  cited={draft.cited_articles} confidence={draft.confidence}")

# examples/m14_sql_codegen.py
"""Pattern 4: code generation with tests as the verifier (text-to-SQL for Maya's questions).

The model writes SQL; nothing trusts it until it (1) is a single read-only
SELECT, (2) runs on a read-only database, and (3) returns the rows the test
expects. Failures go back to the model as feedback, with an attempt ceiling.
"""
from __future__ import annotations

import re
import sqlite3
from collections import Counter

from examples.m14_common import pick_llm
from supportdesk.data import load_tickets
from supportdesk.stand_in import ScriptedLLM

SCHEMA = ("CREATE TABLE tickets (id TEXT PRIMARY KEY, subject TEXT, body TEXT, customer_tier TEXT, "
          "language TEXT, category TEXT, priority TEXT, answerable INTEGER)")


def make_db() -> sqlite3.Connection:
    conn = sqlite3.connect(":memory:")
    conn.execute(SCHEMA)
    conn.executemany("INSERT INTO tickets VALUES (?,?,?,?,?,?,?,?)", [
        (t.id, t.subject, t.body, t.customer_tier, t.language, t.gold["category"], t.gold["priority"],
         int(t.gold["answerable"])) for t in load_tickets()])
    conn.execute("PRAGMA query_only = ON")  # the database itself refuses writes from here on
    return conn


def extract_sql(text: str) -> str:
    fenced = re.search(r"```(?:sql)?\s*(.+?)```", text, re.S)
    return (fenced.group(1) if fenced else text).strip().rstrip(";")


def verify(conn: sqlite3.Connection, sql: str, expected: set[tuple]) -> str:
    """Return '' if the SQL passes, otherwise feedback the model can act on."""
    if not re.fullmatch(r"(?is)\s*(select|with)\b.*", sql) or ";" in sql:
        return "Only a single SELECT statement is allowed."
    try:
        rows = set(conn.execute(sql).fetchall())
    except sqlite3.Error as exc:
        return f"SQLite error: {exc}"
    if rows != expected:
        return f"Wrong result. Got {sorted(rows)[:5]} ({len(rows)} rows); expected {len(expected)} rows."
    return ""


def generate_sql(question: str, llm, conn, expected: set[tuple], max_attempts: int = 3) -> tuple[str, int]:
    messages = [{"role": "system", "content": f"Write one SQLite SELECT for this schema. Reply with SQL only.\n{SCHEMA}\n"
                                              "category values: billing, cancellation, account_access, bug, how_to, "
                                              "feature_request. priority values: low, normal, high, urgent."},
                {"role": "user", "content": question}]
    for attempt in range(1, max_attempts + 1):
        reply = llm(messages, temperature=0.0).text
        sql = extract_sql(reply)
        feedback = verify(conn, sql, expected)
        print(f"  attempt {attempt}: {sql!r}\n    verifier: {feedback or 'PASS'}")
        if not feedback:
            return sql, attempt
        messages += [{"role": "assistant", "content": reply}, {"role": "user", "content": feedback + " Try again."}]
    raise RuntimeError(f"no passing SQL after {max_attempts} attempts")


# STAND-IN replies (hand-written to show each verifier branch; not model output).
SCRIPTED = [
    "DELETE FROM tickets WHERE category = 'billing'",
    "```sql\nSELECT language, COUNT(*) FROM tickets WHERE category = 'Billing' GROUP BY language\n```",
    "```sql\nSELECT language, COUNT(*) FROM tickets WHERE category = 'billing' GROUP BY language;\n```",
]

if __name__ == "__main__":
    conn = make_db()
    # The test: expected rows computed in plain Python from the same data, independently of any SQL.
    expected = set(Counter(t.language for t in load_tickets() if t.gold["category"] == "billing").items())
    print("expected:", sorted(expected))
    llm = pick_llm(ScriptedLLM(replies=list(SCRIPTED)))
    sql, attempts = generate_sql("How many billing tickets per language?", llm, conn, expected)
    print(f"accepted after {attempts} attempts: {sql}")
    try:
        conn.execute("DELETE FROM tickets")
    except sqlite3.OperationalError as exc:
        print("direct write attempt:", exc)

# examples/m14_review_queue.py
"""Pattern 5: content generation with human review (help-center article drafts).

Tickets the help center could not answer are grouped into gaps; the model drafts
an article per gap; drafts enter a review queue whose state machine makes it
impossible to publish anything a named human has not approved.
"""
from __future__ import annotations

import json
import sqlite3
from collections import defaultdict
from datetime import datetime, timezone

from examples.m14_common import pick_llm
from supportdesk.data import load_tickets
from supportdesk.stand_in import ScriptedLLM

TRANSITIONS = {"draft": {"in_review"}, "in_review": {"approved", "changes_requested", "rejected"},
               "changes_requested": {"in_review"}, "approved": {"published"}, "rejected": set(), "published": set()}


class ReviewQueue:
    def __init__(self) -> None:
        self.db = sqlite3.connect(":memory:")
        self.db.execute("CREATE TABLE items (id INTEGER PRIMARY KEY, title TEXT, body TEXT, sources TEXT, "
                        "state TEXT, reviewer TEXT, history TEXT)")

    def add(self, title: str, body: str, sources: list[str]) -> int:
        cur = self.db.execute("INSERT INTO items (title, body, sources, state, reviewer, history) VALUES (?,?,?,?,?,?)",
                              (title, body, json.dumps(sources), "draft", "", "[]"))
        return cur.lastrowid

    def move(self, item_id: int, new_state: str, actor: str, note: str = "") -> None:
        state, history = self.db.execute("SELECT state, history FROM items WHERE id=?", (item_id,)).fetchone()
        if new_state not in TRANSITIONS[state]:
            raise ValueError(f"item {item_id}: {state} -> {new_state} is not allowed")
        if new_state == "approved" and not actor.startswith("human:"):
            raise PermissionError("only a named human reviewer can approve")
        body = self.db.execute("SELECT body FROM items WHERE id=?", (item_id,)).fetchone()[0]
        if new_state == "approved" and "[VERIFY]" in body:
            raise ValueError(f"item {item_id}: resolve every [VERIFY] marker before approving")
        events = json.loads(history) + [{"from": state, "to": new_state, "actor": actor, "note": note,
                                         "at": datetime.now(timezone.utc).isoformat(timespec="seconds")}]
        self.db.execute("UPDATE items SET state=?, reviewer=?, history=? WHERE id=?",
                        (new_state, actor if new_state == "approved" else "", json.dumps(events), item_id))

    def edit(self, item_id: int, body: str, actor: str) -> None:
        self.db.execute("UPDATE items SET body=? WHERE id=?", (body, item_id))
        history = json.loads(self.db.execute("SELECT history FROM items WHERE id=?", (item_id,)).fetchone()[0])
        history.append({"edit_by": actor})
        self.db.execute("UPDATE items SET history=? WHERE id=?", (json.dumps(history), item_id))

    def counts(self) -> dict[str, int]:
        return dict(self.db.execute("SELECT state, COUNT(*) FROM items GROUP BY state ORDER BY state").fetchall())


def draft_messages(category: str, tickets) -> list[dict]:
    asks = "\n".join(f"- [{t.id}] {t.subject}: {t.body}" for t in tickets)
    return [{"role": "system", "content": "Draft a short help-center article (title line, then 3 to 6 sentences) that "
                                          "would answer these customer questions. Mark any fact you are unsure of "
                                          "with [VERIFY]. Never invent prices, dates, or policies."},
            {"role": "user", "content": f"Category: {category}\n{asks}"}]


def responder(messages, kwargs):
    """STAND-IN: a templated draft per gap, not model output."""
    category = messages[-1]["content"].splitlines()[0].split(": ")[1]
    return (f"Help with {category.replace('_', ' ')} requests we cannot resolve automatically\n"
            "Some requests need a person on our team. [VERIFY] Typical reply time is one business day.")


if __name__ == "__main__":
    gaps = defaultdict(list)
    for t in load_tickets():
        if not t.gold["answerable"]:
            gaps[t.gold["category"]].append(t)
    print({c: [t.id for t in ts] for c, ts in gaps.items()})

    llm = pick_llm(ScriptedLLM(responder=responder))
    queue = ReviewQueue()
    ids = {}
    for category, tickets in gaps.items():
        text = llm(draft_messages(category, tickets), temperature=0.3).text
        title, body = text.split("\n", 1)
        ids[category] = queue.add(title, body, [t.id for t in tickets])
        queue.move(ids[category], "in_review", actor="system")
    print("after drafting:", queue.counts())

    try:
        queue.move(ids["billing"], "approved", actor="human:maya")
    except ValueError as exc:
        print("refused:", exc)
    queue.edit(ids["billing"], "Some billing requests need a person on our team. We reply within the times "
               "listed in your plan.", actor="human:maya")
    queue.move(ids["billing"], "approved", actor="human:maya", note="checked against billing policy")
    queue.move(ids["billing"], "published", actor="system")
    queue.move(ids["bug"], "changes_requested", actor="human:maya", note="remove the [VERIFY] reply-time claim")
    queue.move(ids["feature_request"], "rejected", actor="human:maya", note="roadmap questions go to product")
    for attempt in ((ids["account_access"], "approved", "model:auto-reviewer"), (ids["how_to"], "published", "system")):
        try:
            queue.move(*attempt)
        except (ValueError, PermissionError) as exc:
            print("refused:", exc)
    print("final:", queue.counts())
    unverified = queue.db.execute("SELECT COUNT(*) FROM items WHERE state='published' AND body LIKE '%[VERIFY]%'")
    print("published items still containing [VERIFY]:", unverified.fetchone()[0])

# examples/m14_enrich_pipeline.py
"""Pattern 6: data transformation and enrichment (turning raw tickets into analytics rows).

Deterministic fields come from code (regexes, lookups). Only the fuzzy field
(a one-line English gist) comes from the model, cached by a hash of the exact
request so reruns cost nothing, validated, and routed to a dead-letter file
when invalid. The pipeline is idempotent: running it twice gives the same rows.
"""
from __future__ import annotations

import hashlib
import json
import re
from pathlib import Path

from pydantic import BaseModel, Field, ValidationError

from examples.m14_common import pick_llm
from supportdesk.data import Ticket, load_tickets
from supportdesk.stand_in import ScriptedLLM

INVOICE = re.compile(r"\bINV-\d{4}-\d{6}\b")
VAT = re.compile(r"\b(?:DE\d{9}|FR[A-Z0-9]{2}\d{9}|GB\d{9})\b")
PLAN = re.compile(r"\b(Free|Team|Business|Enterprise)\b")
SEATS = re.compile(r"\b(\d{1,5})\s+(?:people|users|seats)\b", re.I)
PROMPT_VERSION = "gist-v1"


class Gist(BaseModel):
    gist_en: str = Field(min_length=8, max_length=140)


class Row(BaseModel):
    ticket_id: str
    invoice_ids: list[str]
    vat_ids: list[str]
    plans_mentioned: list[str]
    seats: int | None
    gist_en: str
    prompt_version: str


def gist_messages(ticket: Ticket) -> list[dict]:
    return [{"role": "system", "content": 'Return JSON {"gist_en": "..."}: the request in at most 15 English words.'},
            {"role": "user", "content": ticket.text}]


class CachedLLM:
    """Content-addressed cache: same model + same messages = same key = no second call."""

    def __init__(self, llm, path: Path) -> None:
        self.llm, self.path = llm, path
        self.store = json.loads(path.read_text()) if path.exists() else {}
        self.hits = self.misses = 0

    def __call__(self, messages: list[dict], **kwargs) -> str:
        key = hashlib.sha256(json.dumps([PROMPT_VERSION, messages, kwargs], sort_keys=True).encode()).hexdigest()
        if key in self.store:
            self.hits += 1
        else:
            self.misses += 1
            self.store[key] = self.llm(messages, **kwargs).text
            self.path.write_text(json.dumps(self.store))
        return self.store[key]


def enrich(ticket: Ticket, cached: CachedLLM) -> Row:
    raw = cached(gist_messages(ticket), temperature=0.0)
    gist = Gist.model_validate_json(raw)  # raises ValidationError; the caller dead-letters it
    seats = SEATS.search(ticket.body)
    return Row(ticket_id=ticket.id, invoice_ids=INVOICE.findall(ticket.text), vat_ids=VAT.findall(ticket.text),
               plans_mentioned=sorted(set(PLAN.findall(ticket.text))), seats=int(seats.group(1)) if seats else None,
               gist_en=gist.gist_en, prompt_version=PROMPT_VERSION)


def run(tickets: list[Ticket], cached: CachedLLM, out: Path, dead: Path) -> tuple[int, int]:
    rows, failures = [], []
    for t in tickets:
        try:
            rows.append(enrich(t, cached).model_dump())
        except ValidationError as exc:
            failures.append({"ticket_id": t.id, "error": exc.errors()[0]["type"]})
    out.write_text("".join(json.dumps(r, ensure_ascii=False) + "\n" for r in rows))   # overwrite, never append
    dead.write_text("".join(json.dumps(f) + "\n" for f in failures))
    return len(rows), len(failures)


def responder(messages, kwargs):
    """STAND-IN: uses the subject as the gist; Japanese subjects come back too short, to exercise dead-lettering."""
    subject = messages[-1]["content"].splitlines()[0].removeprefix("Subject: ")
    return json.dumps({"gist_en": subject if subject.isascii() else "?"})


if __name__ == "__main__":
    work = Path("m14_out")
    work.mkdir(exist_ok=True)
    cache_file = work / "gist_cache.json"
    cache_file.unlink(missing_ok=True)
    cached = CachedLLM(pick_llm(ScriptedLLM(responder=responder)), cache_file)
    tickets = load_tickets()
    for run_no in (1, 2):
        before = (cached.hits, cached.misses)
        ok, bad = run(tickets, cached, work / "enriched.jsonl", work / "dead_letter.jsonl")
        print(f"run {run_no}: rows {ok}, dead-letter {bad}, cache hits {cached.hits - before[0]}, "
              f"model calls {cached.misses - before[1]}")
    rows = [json.loads(line) for line in (work / "enriched.jsonl").read_text().splitlines()]
    for r in rows:
        if r["invoice_ids"] or r["vat_ids"] or r["seats"]:
            print(r)
    print("dead-letter:", (work / "dead_letter.jsonl").read_text().strip().replace("\n", " | "))

# examples/m14_ticket_search.py
"""Pattern 7: search and question answering over a private corpus (past tickets).

Agents ask "have we seen this before?". We index past (dev) tickets with the same
BM25 class the help center uses, filter by permission BEFORE ranking, and
measure how often the most similar past ticket shares the new ticket's category.
"""
from __future__ import annotations

from examples.m14_common import fmt_rate
from supportdesk.data import Article, Ticket, load_tickets
from supportdesk.kb_search import KBSearch

# Who may read what: Enterprise tickets can contain contract terms, so only the enterprise pod sees them.
ACL = {"free": {"all-agents"}, "team": {"all-agents"}, "business": {"all-agents"}, "enterprise": {"enterprise-pod"}}


def as_article(t: Ticket) -> Article:
    return Article(id=t.id, title=t.subject, tags=(t.customer_tier,), body=t.body)


class TicketSearch:
    def __init__(self, past: list[Ticket]) -> None:
        self.past = {t.id: t for t in past}
        self._indexes: dict[frozenset[str], KBSearch] = {}

    def _index_for(self, groups: frozenset[str]) -> KBSearch:
        """One BM25 index per permission set, built only from tickets those groups may read.

        Filtering before ranking matters twice: restricted text never reaches the results, and it
        never influences the word statistics (IDF) that score everyone else's results either.
        """
        if groups not in self._indexes:
            visible = [as_article(t) for t in self.past.values() if ACL[t.customer_tier] & groups]
            self._indexes[groups] = KBSearch(visible)
        return self._indexes[groups]

    def search(self, query: str, groups: set[str], k: int = 3):
        return self._index_for(frozenset(groups)).search(query, k)


if __name__ == "__main__":
    dev, test = load_tickets("dev"), load_tickets("test")
    search = TicketSearch(dev)
    agree = found = 0
    for t in test:
        hits = search.search(t.text, {"all-agents", "enterprise-pod"}, k=1)
        if hits:
            found += 1
            agree += search.past[hits[0].article_id].gold["category"] == t.gold["category"]
    print(f"test tickets with any similar past ticket: {found}/{len(test)}")
    print(f"top-1 past ticket has the same category: {fmt_rate(agree, found)}")

    query = "custom SAML login domain for our Okta users"
    for groups in ({"all-agents"}, {"all-agents", "enterprise-pod"}):
        print(sorted(groups), [(h.article_id, h.title, h.score) for h in search.search(query, groups)])