Part B: Design Principles

# examples/m14_autonomy.py
"""Design principles: match autonomy to consequence, and design for the model being wrong.

Autonomy is derived from properties of the action (can it be undone, does the
customer see it, does it move money or data), not decided case by case. Every
automatic action is logged with its inverse so a human can undo it, and a model
failure falls back to a safe default instead of blocking the queue.
"""
from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass, field

from pydantic import ValidationError

from supportdesk.schemas import Triage
from supportdesk.stand_in import ScriptedLLM

LEVELS = ("auto", "auto_then_notify", "suggest_and_approve", "human_only")


@dataclass(frozen=True)
class Action:
    name: str
    reversible: bool          # can we fully undo it within minutes?
    customer_visible: bool    # does the customer see or receive it?
    money_or_data: bool       # does it move money, delete data, or change access?
    blast_radius: int         # how many customers one mistake can reach


def autonomy_for(a: Action) -> str:
    if a.money_or_data and not a.reversible:
        return "human_only"
    if a.money_or_data or a.customer_visible or a.blast_radius > 1:
        return "suggest_and_approve"
    if not a.reversible:
        return "auto_then_notify"
    return "auto"


ACTIONS = [
    Action("add_category_tag", True, False, False, 1),
    Action("set_priority", True, False, False, 1),
    Action("route_to_queue", True, False, False, 1),
    Action("post_internal_note", False, False, False, 1),
    Action("send_reply_with_kb_link", False, True, False, 1),
    Action("send_bulk_incident_update", False, True, False, 500),
    Action("unlock_account", True, True, True, 1),
    Action("issue_refund", False, True, True, 1),
    Action("delete_personal_data", False, True, True, 1),
]


@dataclass
class Ticketish:
    id: str
    tags: set[str] = field(default_factory=set)
    priority: str = "normal"


class ActionLog:
    """Every automatic change is stored with the operation that reverses it."""

    def __init__(self) -> None:
        self.entries: list[tuple[str, str, Callable[[], None]]] = []

    def apply(self, ticket: Ticketish, name: str, do, undo) -> None:
        do()
        self.entries.append((ticket.id, name, undo))

    def undo_last(self, ticket_id: str) -> str:
        for i in range(len(self.entries) - 1, -1, -1):
            if self.entries[i][0] == ticket_id:
                _, name, undo = self.entries.pop(i)
                undo()
                return name
        return "nothing to undo"


def triage_or_fallback(llm, text: str) -> tuple[Triage | None, str]:
    """If the model call fails or returns junk, degrade to 'normal priority, human queue', never block."""
    try:
        return Triage.model_validate_json(llm([{"role": "user", "content": text}]).text), "model"
    except (ValidationError, RuntimeError, TimeoutError) as exc:
        return None, f"fallback ({type(exc).__name__}): route to human queue at normal priority"


if __name__ == "__main__":
    print(f"{'action':28} {'rev':>4} {'seen':>5} {'$/data':>6} {'reach':>6}  autonomy")
    for a in ACTIONS:
        print(f"{a.name:28} {str(a.reversible)[0]:>4} {str(a.customer_visible)[0]:>5} "
              f"{str(a.money_or_data)[0]:>6} {a.blast_radius:>6}  {autonomy_for(a)}")

    log, ticket = ActionLog(), Ticketish("T-1001")
    old = ticket.priority
    log.apply(ticket, "add_category_tag", lambda: ticket.tags.add("billing"), lambda: ticket.tags.discard("billing"))
    log.apply(ticket, "set_priority", lambda: setattr(ticket, "priority", "low"),
              lambda: setattr(ticket, "priority", old))
    print("\nafter model actions:", ticket)
    print("agent clicks undo:", log.undo_last("T-1001"), "->", ticket)

    # STAND-IN replies: one valid, one truncated, then an outage (not model output).
    replies = ['{"category": "billing", "priority": "high", "language": "en", "summary": "Duplicate charge.", '
               '"needs_human": true}', '{"category": "billing", "prio']
    llm = ScriptedLLM(replies=replies)
    for _ in range(3):
        triage, source = triage_or_fallback(llm, "Charged twice this month")
        print(source if triage is None else f"{source}: {triage.category}/{triage.priority}")

# examples/m14_reviewer_fatigue.py
"""Design principles: where to place human review, and what reviewer fatigue does to it.

ASSUMPTION (not measured, replace with your own audit data): the chance a reviewer
catches a wrong draft decays with the number of items already reviewed in the
session, p(i) = floor + (start - floor) * exp(-i / tau). The math that follows
from the assumption is exact, and a seeded Monte Carlo run checks it.
"""
from __future__ import annotations

import math

import numpy as np

from examples.m14_common import fmt_rate

START, FLOOR, TAU = 0.95, 0.60, 40.0   # ASSUMED reviewer behaviour
ERROR_RATE = 0.08                       # ASSUMED share of drafts that are wrong
DAILY = 300                             # drafts per day needing a decision


def catch_prob(i: int) -> float:
    return FLOOR + (START - FLOOR) * math.exp(-i / TAU)


def expected_escapes(n_items: int, error_rate: float, session: int) -> float:
    """Expected wrong drafts that get through when n_items are reviewed in sessions of `session` items."""
    return sum(error_rate * (1 - catch_prob(i % session)) for i in range(n_items))


def simulate(n_items: int, error_rate: float, session: int, days: int, seed: int = 14) -> float:
    rng = np.random.default_rng(seed)
    p = np.array([catch_prob(i % session) for i in range(n_items)])
    wrong = rng.random((days, n_items)) < error_rate
    caught = rng.random((days, n_items)) < p
    return float((wrong & ~caught).sum(axis=1).mean())


if __name__ == "__main__":
    print("session length vs average catch rate (under the assumption)")
    for n in (10, 25, 50, 100, 200, 300):
        avg = sum(catch_prob(i) for i in range(n)) / n
        print(f"  {n:>3} items in one sitting: average catch {avg:.1%}, last item {catch_prob(n - 1):.1%}")

    print(f"\n{DAILY} drafts/day, {ERROR_RATE:.0%} wrong = {DAILY * ERROR_RATE:.0f} wrong drafts/day")
    plans = {
        "A review all, one sitting": (DAILY, ERROR_RATE, DAILY, 0.0),
        "B review all, breaks every 40": (DAILY, ERROR_RATE, 40, 0.0),
    }
    # C and D: review only drafts flagged low-confidence. ASSUMED: 30% are flagged, holding 70% or 90% of errors.
    flagged = int(DAILY * 0.30)
    for label, share in (("C", 0.70), ("D", 0.90)):
        rate = DAILY * ERROR_RATE * share / flagged
        plans[f"{label} review flagged 30% ({share:.0%} of errors)"] = (flagged, rate, 40, DAILY * ERROR_RATE * (1 - share))
    for name, (n, rate, session, unreviewed) in plans.items():
        exact = expected_escapes(n, rate, session) + unreviewed
        sim = simulate(n, rate, session, days=20_000) + unreviewed
        print(f"  {name:40} reviewed {n:>3}  escapes/day exact {exact:5.2f}  simulated {sim:5.2f}")

    print("\nmeasuring the real curve: plant known-wrong drafts and count catches")
    for caught, planted in ((17, 20), (85, 100)):
        print(f"  caught {fmt_rate(caught, planted)}")

# examples/m14_draft_card.py
"""Design principles: trust calibration and progressive disclosure in the interface.

The agent sidebar shows WHY to trust a draft (sources, and a confidence label
whose meaning was measured), not just the draft. The customer sees a disclosure
line that matches how much the AI did.
"""
from __future__ import annotations

from examples.m14_common import fmt_rate
from supportdesk.data import get_article, load_tickets
from supportdesk.kb_search import KBSearch
from supportdesk.schemas import DraftReply

KB = KBSearch()
BANDS = [(12.0, "strong match"), (6.5, "partial match"), (0.0, "weak match")]


def band(score: float) -> str:
    return next(label for floor, label in BANDS if score >= floor)


def calibration_table(tickets) -> dict[str, tuple[int, int]]:
    """How often the top article is the right one, per displayed band (measured, answerable tickets only)."""
    table = {label: [0, 0] for _, label in BANDS}
    for t in tickets:
        hits = KB.search(t.text, 1)
        if t.gold["answerable"] and hits:
            cell = table[band(hits[0].score)]
            cell[0] += hits[0].article_id == t.gold["kb_article"]
            cell[1] += 1
    return {k: (v[0], v[1]) for k, v in table.items()}


def agent_card(ticket_id: str, draft: DraftReply, top_score: float, table) -> str:
    right, n = table[band(top_score)]
    lines = [f"AI DRAFT for {ticket_id}  (review before sending)",
             f"Source match: {band(top_score)}; historically the right article {right} of {n} times",
             "Sources:"]
    lines += [f"  [{a}] {get_article(a).title}" for a in draft.cited_articles] or ["  none: treat as unsupported"]
    lines += ["Draft:", f"  {draft.reply}", "[Send]  [Edit]  [Discard]  [Wrong source]"]
    return "\n".join(lines)


DISCLOSURE = {
    "human_wrote": "",
    "ai_drafted_human_sent": "This reply was drafted with AI assistance and reviewed by {agent}.",
    "ai_sent": "This is an automated answer from our help center. Reply 'agent' to reach a person.",
}


def customer_footer(mode: str, agent: str = "Maya") -> str:
    return DISCLOSURE[mode].format(agent=agent)


if __name__ == "__main__":
    table = calibration_table(load_tickets("dev"))
    for label, (right, n) in table.items():
        print(f"{label:14} top article correct: {fmt_rate(right, n) if n else 'no tickets'}")

    ticket = next(t for t in load_tickets("test") if t.id == "T-1048")
    top = KB.search(ticket.text, 1)[0]
    draft = DraftReply(reply="After 5 failed sign-in attempts an account locks for 15 minutes. For security "
                             "reasons our team cannot unlock it sooner.",
                       cited_articles=[top.article_id], confidence="medium")  # hand-written example draft
    print("\n" + agent_card(ticket.id, draft, top.score, table))
    for mode in DISCLOSURE:
        print(f"\nfooter[{mode}]: {customer_footer(mode) or '(none)'}")

# examples/m14_feedback_events.py
"""Design principles: capture feedback from day one with a fixed event schema.

Every draft shown to an agent produces events. Because each event carries the
prompt version and model, next month's question "did v2 help?" is a query, not
an archaeology project. Edit size is measured, not self-reported.
"""
from __future__ import annotations

import difflib
import hashlib
import json
import random
import uuid
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Literal

from pydantic import BaseModel, Field

from examples.m14_common import fmt_rate

EventType = Literal["draft_shown", "draft_sent", "draft_discarded", "thumbs_up", "thumbs_down", "escalated",
                    "wrong_source_flagged", "undo"]


class FeedbackEvent(BaseModel):
    event_id: str = Field(default_factory=lambda: uuid.uuid4().hex)
    at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat(timespec="seconds"))
    event: EventType
    ticket_id: str
    draft_id: str
    feature: Literal["triage", "draft_reply", "kb_article_draft"]
    prompt_version: str
    model: str
    agent: str                              # pseudonymous id, never a name or email
    edit_ratio: float | None = None         # 0 = sent unchanged, 1 = fully rewritten
    cited_articles: list[str] = []
    cost_usd: float | None = None
    latency_ms: float | None = None


def pseudonym(agent_email: str, salt: str = "rotate-me-quarterly") -> str:
    return "agent_" + hashlib.sha256((salt + agent_email).encode()).hexdigest()[:10]


def edit_ratio(draft: str, sent: str) -> float:
    """1 minus difflib's similarity: how much of the draft the agent changed."""
    return round(1 - difflib.SequenceMatcher(None, draft, sent).ratio(), 3)


def log(event: FeedbackEvent, path: Path) -> None:
    with path.open("a", encoding="utf-8") as f:
        f.write(event.model_dump_json() + "\n")


def summarize(path: Path) -> dict[str, dict[str, float]]:
    by_version: dict[str, list[FeedbackEvent]] = defaultdict(list)
    for line in path.read_text().splitlines():
        e = FeedbackEvent.model_validate_json(line)
        by_version[e.prompt_version].append(e)
    report = {}
    for version, events in sorted(by_version.items()):
        shown = sum(e.event == "draft_shown" for e in events)
        sent = [e for e in events if e.event == "draft_sent"]
        unchanged = sum(e.edit_ratio is not None and e.edit_ratio < 0.05 for e in sent)
        report[version] = {"shown": shown, "sent": len(sent), "sent_unchanged": unchanged,
                           "discarded": sum(e.event == "draft_discarded" for e in events),
                           "mean_edit_ratio": round(sum(e.edit_ratio for e in sent) / max(len(sent), 1), 3)}
    return report


if __name__ == "__main__":
    draft = "After 5 failed sign-in attempts an account locks for 15 minutes."
    for sent in (draft, draft + " Sorry for the trouble!",
                 "Your account locks for 15 minutes after 5 failed attempts; please wait and try again."):
        print(f"edit_ratio={edit_ratio(draft, sent):.3f}  sent: {sent}")

    # SIMULATED traffic (seeded random, NOT real agent behaviour) just to exercise the pipeline end to end.
    path = Path("m14_out/feedback.jsonl")
    path.parent.mkdir(exist_ok=True)
    path.unlink(missing_ok=True)
    rng = random.Random(14)
    agent = pseudonym("maya@brightlane.example")
    for version, p_discard, typical_edit in (("draft-v1", 0.30, 0.35), ("draft-v2", 0.15, 0.20)):
        for i in range(40):
            common = dict(ticket_id=f"T-{2000 + i}", draft_id=f"{version}-{i}", feature="draft_reply",
                          prompt_version=version, model="openai/gpt-oss-120b", agent=agent)
            log(FeedbackEvent(event="draft_shown", **common), path)
            if rng.random() < p_discard:
                log(FeedbackEvent(event="draft_discarded", **common), path)
            else:
                ratio = max(0.0, min(1.0, rng.gauss(typical_edit, 0.15)))
                log(FeedbackEvent(event="draft_sent", edit_ratio=round(ratio, 3), **common), path)
    print(f"\n{sum(1 for _ in path.open())} events written for agent {agent}")
    report = summarize(path)
    print(json.dumps(report, indent=1))
    for version, r in report.items():
        print(f"{version} discard rate {fmt_rate(r['discarded'], r['shown'])}")