নির্ভরযোগ্য আউটপুট চুক্তি

# file: evidencedesk/contracts.py
"""Output contracts: versions, front-end states, event rules, bounds, untrusted values, adapters."""
import json
import unicodedata
from collections.abc import Iterable, Iterator
from pathlib import Path
from typing import Annotated, Literal
from urllib.parse import urlsplit

from pydantic import BaseModel, ConfigDict, Field, StringConstraints, ValidationError

from evidencedesk.llm import LLMRequest, LLMResponse, Message, Usage
from evidencedesk.results import SORRY, AnswerResult, Outcome
from evidencedesk.streaming import Assembled, StreamEvent, assemble, safe_events

MAX_BYTES = 4000  # one structured reply
MAX_CHARS = 4000  # one streamed answer
HOSTS = frozenset({"help.kestrelcycles.example"})
NO_ANSWER = "No answer was produced. Please rephrase the question."
FAILURES = Path(__file__).resolve().parent.parent / "fixtures" / "failures.jsonl"
RETRY_LATER = {"invalid_json", "invalid_schema", "interrupted", "stream_error", "untrusted_value"}
GOOD = {"schema_version": 2, "answer": "30 days.", "cited_ids": ["kc-returns-v3"], "confidence": "high", "status": "answered"}
Ident = Annotated[str, StringConstraints(pattern=r"^[a-z0-9][a-z0-9#._-]{0,63}$")]


class Answer(AnswerResult):
    cited_ids: list[Ident] = Field(max_length=5)
    links: list[Annotated[str, StringConstraints(max_length=200)]] = Field(default_factory=list, max_length=5)


def read_answer(payload: object) -> Outcome:
    """Accepts v1 and v2, rejects everything else instead of guessing."""
    version = payload.get("schema_version") if isinstance(payload, dict) else None
    if type(version) is not int or version not in (1, 2):
        return Outcome(False, "unknown_version")  # also for a reply that is not an object
    body = {k: v for k, v in payload.items() if k != "schema_version"}
    try:
        if version == 1:
            body = AnswerResult.model_validate(body).model_dump()  # the old rules, unchanged
        return Outcome(True, "ok", Answer.model_validate(body))
    except ValidationError as exc:
        found = tuple({"loc": ".".join(map(str, e["loc"])), "msg": e["msg"]} for e in exc.errors())
        return Outcome(False, "invalid_schema", problems=found)


def parse_contract(response: LLMResponse, max_bytes: int = MAX_BYTES) -> Outcome:
    if response.stop_reason != "end":
        return Outcome(False, "truncated" if response.stop_reason == "length" else "unusable_stop")
    if len(response.text.encode("utf-8")) > max_bytes:
        return Outcome(False, "too_large")  # reject, do not cut
    try:
        payload = json.loads(response.text)
    except (ValueError, RecursionError):
        return Outcome(False, "invalid_json")
    return read_answer(payload)

from evidencedesk.contracts import read_answer

v1 = {"schema_version": 1, "answer": "You can return a new bike within 30 days.", "cited_ids": ["kc-returns-v3"],
      "confidence": "high", "status": "answered"}
v2 = {**v1, "schema_version": 2, "links": ["https://help.kestrelcycles.example/returns"]}
cases = {"v1": v1, "v2": v2, "v3 (future)": {**v2, "schema_version": 3},
         "v1 with links": {**v1, "links": []}}
for name, payload in cases.items():
    outcome = read_answer(payload)
    print(f"{name:20} {outcome.code:16} links={outcome.value.links if outcome.ok else '-'}")

v1                   ok               links=[]
v2                   ok               links=['https://help.kestrelcycles.example/returns']
v3 (future)          unknown_version  links=-
v1 with links        invalid_schema   links=-

class Envelope(BaseModel):
    model_config = ConfigDict(extra="forbid")
    schema_version: Literal[2] = 2
    state: Literal["ok", "empty", "partial", "error"]
    text: str
    code: str = "ok"
    retryable: bool = False
    result: Answer | None = None


def to_envelope(source: Outcome | Assembled) -> Envelope:
    if isinstance(source, Outcome):
        if source.ok:
            return Envelope(state="ok", text=source.value.answer, result=Answer(**source.value.model_dump()))
        return Envelope(state="error", text=SORRY, code=source.code, retryable=source.code in RETRY_LATER)
    if source.status == "empty":
        return Envelope(state="empty", text=NO_ANSWER, code="empty")
    if source.status == "interrupted":
        return Envelope(state="partial", text=source.text, code="interrupted", retryable=True)
    if source.status == "error":
        return Envelope(state="error", text=SORRY, code="stream_error", retryable=True)
    if source.stop_reason == "length":
        return Envelope(state="partial", text=source.text, code="truncated")
    return Envelope(state="ok", text=source.text)

from evidencedesk.contracts import to_envelope
from evidencedesk.results import Outcome
from evidencedesk.streaming import Assembled

sources = {"valid structured": read_answer(v2), "empty stream": Assembled("", "empty"),
           "interrupted": Assembled("Standard delivery ta", "interrupted"),
           "cut off (length)": Assembled("For storage longer than 30", "complete", stop_reason="length"),
           "bad JSON": Outcome(False, "invalid_json")}
for name, source in sources.items():
    e = to_envelope(source)
    print(f"{name:17} {e.state:8} {e.code:12} retry={e.retryable!s:5} {e.text[:28]!r}")

valid structured  ok       ok           retry=False 'You can return a new bike wi'
empty stream      empty    empty        retry=False 'No answer was produced. Plea'
interrupted       partial  interrupted  retry=True  'Standard delivery ta'
cut off (length)  partial  truncated    retry=False 'For storage longer than 30'
bad JSON          error    invalid_json retry=True  'Sorry, I could not produce a'

def check_events(events: Iterable[StreamEvent]) -> list[str]:
    problems, ended = [], False
    for n, event in enumerate(events):
        if event.kind not in ("chunk", "done", "error"):
            problems.append(f"event {n}: unknown kind {event.kind!r}")
        elif ended:
            problems.append(f"event {n}: {event.kind} after the stream ended")
        elif event.kind == "chunk" and not event.text:
            problems.append(f"event {n}: empty chunk")
        elif event.kind != "chunk":
            ended = True
    return problems + ([] if ended else ["no terminal event (done or error)"])

from evidencedesk.contracts import check_events
from evidencedesk.llm import LLMRequest, Message
from evidencedesk.streaming import FakeStreamClient, StreamEvent

request = LLMRequest((Message("user", "How long does standard delivery take?"),))
good = list(FakeStreamClient(chunk_size=30).stream(request))
cases = {"good": good, "interrupted": good[:2], "chunk after done": good + [StreamEvent("chunk", "x")],
         "error after done": good + [StreamEvent("error", "late")], "empty chunk": [StreamEvent("chunk", ""), good[-1]]}
for name, events in cases.items():
    print(f"{name:17}", check_events(events) or "ok")

good              ok
interrupted       ['no terminal event (done or error)']
chunk after done  ['event 4: chunk after the stream ended']
error after done  ['event 4: error after the stream ended']
empty chunk       ['event 0: empty chunk']

def watched(events: Iterable[StreamEvent], log: list[str]) -> Iterator[StreamEvent]:
    try:
        for event in events:
            log.append(event.kind)
            yield event
    finally:
        log.append("closed")


def bounded(events: Iterable[StreamEvent], max_chars: int = MAX_CHARS) -> Iterator[StreamEvent]:
    """Reject, do not cut: past the limit send one error event, stop reading and close the source."""
    total = 0
    try:
        for event in events:
            total += len(event.text) if event.kind == "chunk" else 0
            if total > max_chars:
                yield StreamEvent("error", "too_large")
                return
            yield event
    finally:
        getattr(events, "close", lambda: None)()

from evidencedesk.contracts import bounded, watched

log = []
events = watched(FakeStreamClient(chunk_size=10).stream(request), log)
next(events), next(events)
print("after 2 reads:", log)
events.close()
print("after close():", log)
log = []
kept = list(bounded(watched(FakeStreamClient(chunk_size=10).stream(request), log), max_chars=25))
print("bounded:", [e.kind for e in kept], "| source log:", log)

after 2 reads: ['chunk', 'chunk']
after close(): ['chunk', 'chunk', 'closed']
bounded: ['chunk', 'chunk', 'error'] | source log: ['chunk', 'chunk', 'chunk', 'closed']

def clusters(text: str) -> list[str]:
    """Approximate grapheme clusters: a base plus marks, joiners and virama joins."""
    out: list[str] = []
    for ch in text:
        last = out[-1][-1] if out else ""
        mark = unicodedata.category(ch)[0] == "M" or ch in "\u200c\u200d"
        if last and (mark or unicodedata.combining(last) == 9 or last == "\u200d"):
            out[-1] += ch
        else:
            out.append(ch)
    return out


def chunk_clusters(text: str, size: int) -> list[str]:
    chunks, current = [], ""
    for cluster in clusters(text):
        if current and len(current) + len(cluster) > size:
            chunks.append(current)
            current = ""
        current += cluster
    return chunks + [current] if current else chunks


def broken_boundaries(chunks: list[str]) -> int:
    return sum(len(clusters(a + b)) < len(clusters(a)) + len(clusters(b)) for a, b in zip(chunks, chunks[1:]))

import unicodedata
from evidencedesk.contracts import broken_boundaries, chunk_clusters, clusters
from evidencedesk.streaming import chunk_text

texts = {"Bangla": "কিংবা ক্ষমা কার্যদিবস", "e + accent": "cafe\u0301 e\u0301te\u0301"}
for name, text in texts.items():
    print(f"{name}: {len(text)} code points, {len(clusters(text))} clusters")
    for size in (1, 3):
        plain, safe = chunk_text(text, size), chunk_clusters(text, size)
        print(f"  size {size}: by characters broken={broken_boundaries(plain)}, by clusters broken={broken_boundaries(safe)},"
              f" lossless={''.join(plain) == text == ''.join(safe)}")
pieces = chunk_text(texts["Bangla"], 1)[:6]
print(pieces)
print([unicodedata.name(p) for p in pieces if unicodedata.category(p)[0] == "M"])

Bangla: 21 code points, 11 clusters
  size 1: by characters broken=10, by clusters broken=0, lossless=True
  size 3: by characters broken=2, by clusters broken=0, lossless=True
e + accent: 11 code points, 8 clusters
  size 1: by characters broken=3, by clusters broken=0, lossless=True
  size 3: by characters broken=0, by clusters broken=0, lossless=True
['ক', 'ি', 'ং', 'ব', 'া', ' ']
['BENGALI VOWEL SIGN I', 'BENGALI SIGN ANUSVARA', 'BENGALI VOWEL SIGN AA']

def check_url(url: str, hosts: frozenset[str] = HOSTS) -> str | None:
    if not url or any(c.isspace() or not c.isprintable() for c in url):
        return "empty, or has whitespace or control characters"
    try:
        parts = urlsplit(url)
        host = parts.hostname
    except ValueError:
        return "not a valid URL"
    if parts.scheme != "https":
        return f"scheme {parts.scheme or '(none)'!r} is not https"
    if parts.username or parts.password:
        return "credentials in the URL"
    return None if host in hosts else f"host {host!r} is not allowed"


def untrusted_problems(answer: Answer, retrieved: Iterable[str], hosts: frozenset[str] = HOSTS) -> list[str]:
    """Ids must be ones retrieved for THIS question; links must be on our hosts."""
    known = set(retrieved)
    found = [f"cited id {i!r} was not retrieved" for i in answer.cited_ids if i not in known]
    return found + [f"link {u[:30]!r}: {why}" for u in answer.links if (why := check_url(u, hosts))]

from evidencedesk.contracts import check_url, untrusted_problems

urls = ["https://help.kestrelcycles.example/returns", "http://help.kestrelcycles.example/returns", "javascript:alert(1)",
        "https://help.kestrelcycles.example.evil.test/", "https://help.kestrelcycles.example@evil.test/"]
for url in urls:
    print(f"{url:50} {check_url(url) or 'allowed'}")
answer = read_answer({**v2, "cited_ids": ["kc-returns-v3", "kc-privacy-v1"], "links": ["http://evil.test/x"]}).value
for problem in untrusted_problems(answer, retrieved={"kc-returns-v3"}):
    print(problem)

https://help.kestrelcycles.example/returns         allowed
http://help.kestrelcycles.example/returns          scheme 'http' is not https
javascript:alert(1)                                scheme 'javascript' is not https
https://help.kestrelcycles.example.evil.test/      host 'help.kestrelcycles.example.evil.test' is not allowed
https://help.kestrelcycles.example@evil.test/      credentials in the URL
cited id 'kc-privacy-v1' was not retrieved
link 'http://evil.test/x': scheme 'http' is not https

def from_shape_a(raw: dict) -> LLMResponse:
    # hand-written shape A: output_text, status, usage
    cut = raw.get("incomplete_details", {}).get("reason") == "max_output_tokens"
    stop = "end" if raw.get("status") == "completed" else "length" if cut else "other"
    return LLMResponse(raw.get("output_text", ""), "shape-a", Usage(**raw.get("usage", {})), stop)


def from_shape_b(raw: dict) -> LLMResponse:
    # hand-written shape B: content list, stop_reason, tokens (a refusal becomes "other")
    text = "".join(b.get("text", "") for b in raw.get("content", []) if b.get("type") == "text")
    stop = {"end_turn": "end", "max_tokens": "length"}.get(raw.get("stop_reason"), "other")
    t = raw.get("tokens", {})
    return LLMResponse(text, "shape-b", Usage(t.get("prompt", 0), t.get("completion", 0)), stop)

import json
from evidencedesk.contracts import from_shape_a, from_shape_b, parse_contract

body = json.dumps(v2)
# hand-written and simplified: these two dicts are NOT copies of any real provider response
shape_a = {"status": "completed", "output_text": body, "usage": {"input_tokens": 41, "output_tokens": 19}}
shape_b = {"content": [{"type": "text", "text": body}], "stop_reason": "end_turn", "tokens": {"prompt": 41, "completion": 19}}
a, b = from_shape_a(shape_a), from_shape_b(shape_b)
print("same usage:", a.usage == b.usage, "| same value:", parse_contract(a).value == parse_contract(b).value)
cut_a = {**shape_a, "status": "incomplete", "incomplete_details": {"reason": "max_output_tokens"}}
for name, response in {"a: cut off": from_shape_a(cut_a), "b: refusal": from_shape_b({**shape_b, "stop_reason": "refusal"})}.items():
    print(f"{name:11} stop={response.stop_reason:7} -> {parse_contract(response).code}")

same usage: True | same value: True
a: cut off  stop=length  -> truncated
b: refusal  stop=other   -> unusable_stop

{"name": "v1 good", "patch": {"schema_version": 1}, "expect": "ok"}
{"name": "future version", "patch": {"schema_version": 3}, "expect": "unknown_version"}
{"name": "prose, not JSON", "raw": "Sure! It is 30 days.", "expect": "invalid_json"}
{"name": "cut mid-string, stop says end", "raw": "{\"schema_version\": 2, \"answer\": \"30 da", "expect": "invalid_json"}
{"name": "cut off, stop says length", "raw": "{\"schema_version\": 2, \"answer\": \"30 da", "stop": "length", "expect": "truncated"}
{"name": "refusal-like stop", "raw": "", "stop": "other", "expect": "unusable_stop"}
{"name": "unexpected extra key", "patch": {"debug": true}, "expect": "invalid_schema"}
{"name": "path-like id", "patch": {"cited_ids": ["../secret"]}, "expect": "invalid_schema"}
{"name": "reply over the byte limit", "patch": {}, "max_bytes": 60, "expect": "too_large"}

def load_failures(path: Path = FAILURES) -> list[dict]:
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]


def replay(case: dict) -> str:
    """A case has `raw` text or a `patch` on GOOD, plus optional `stop` and `max_bytes`."""
    text = case["raw"] if "raw" in case else json.dumps({**GOOD, **case["patch"]})
    response = LLMResponse(text, "recorded", Usage(), case.get("stop", "end"))
    return parse_contract(response, case.get("max_bytes", MAX_BYTES)).code

def contract_from_stream(streamed: Assembled, retrieved: Iterable[str]) -> Envelope:
    """Half a JSON text is never shown, so a broken structured stream is an error."""
    if streamed.status == "empty":
        return to_envelope(streamed)
    if streamed.status != "complete":
        code = "too_large" if streamed.error == "too_large" else streamed.status.replace("error", "stream_error")
        return to_envelope(Outcome(False, code))
    outcome = parse_contract(LLMResponse(streamed.text, "stream", streamed.usage or Usage(), streamed.stop_reason or "other"))
    if outcome.ok and (found := untrusted_problems(outcome.value, retrieved)):
        outcome = Outcome(False, "untrusted_value", problems=tuple({"loc": "", "msg": f} for f in found))
    return to_envelope(outcome)


def envelope_sse(envelope: Envelope) -> Iterator[str]:
    if envelope.state in ("ok", "partial"):
        for piece in chunk_clusters(envelope.text, 16):
            yield f"event: chunk\ndata: {json.dumps({'text': piece}, ensure_ascii=False)}\n\n"
    kind = "done" if envelope.state in ("ok", "empty") else "error"
    yield f"event: {kind}\ndata: {envelope.model_dump_json()}\n\n"


def make_faq_app(client, retrieved: Iterable[str], max_chars: int = MAX_CHARS):
    from fastapi import FastAPI, Query
    from fastapi.responses import StreamingResponse

    app, known = FastAPI(), frozenset(retrieved)

    @app.get("/faq")
    def faq(q: str = Query(min_length=1, max_length=500)) -> StreamingResponse:
        events = bounded(safe_events(client, LLMRequest((Message("user", q),))), max_chars)
        envelope = contract_from_stream(assemble(events), known)
        return StreamingResponse(envelope_sse(envelope), media_type="text/event-stream")

    return app

from fastapi.testclient import TestClient
from evidencedesk.contracts import check_events, make_faq_app
from evidencedesk.streaming import parse_sse

Q = "How long do I have to return a new bike?"
bn = {**v2, "answer": "নতুন বাইক ৩০ দিনের মধ্যে ফেরত দেওয়া যায়।", "cited_ids": ["kc-returns-v3-bn"]}
lab = [("good v2", v2, {}), ("Bangla answer", bn, {}), ("id not retrieved", {**v2, "cited_ids": ["kc-privacy-v1"]}, {}),
       ("dropped stream", v2, {"stop_after": 2}), ("error mid-stream", v2, {"fail_after": 2})]
print(f"{'case':18} status chunks last  {'state/code':22} check")
for name, payload, options in lab:
    fixtures = {Q: {"text": json.dumps(payload, ensure_ascii=False), "usage": {}, "stop_reason": "end"}}
    client = TestClient(make_faq_app(FakeStreamClient(fixtures, chunk_size=9, **options), {"kc-returns-v3", "kc-returns-v3-bn"}))
    with client.stream("GET", "/faq", params={"q": Q}) as response:
        lines = list(response.iter_lines())
    events = parse_sse(lines)
    final = json.loads([ln for ln in lines if ln.startswith("data: ")][-1][6:])  # the envelope in the terminal event
    print(f"{name:18} {response.status_code:6} {sum(e.kind == 'chunk' for e in events):6} {events[-1].kind:5}"
          f" {final['state'] + '/' + final['code']:22} {check_events(events) or 'ok'}")

case               status chunks last  state/code             check
good v2               200      3 done  ok/ok                  ok
Bangla answer         200      3 done  ok/ok                  ok
id not retrieved      200      0 error error/untrusted_value  ok
dropped stream        200      0 error error/interrupted      ok
error mid-stream      200      0 error error/stream_error     ok

python -m pytest -q tests/test_contracts.py | sed 's/ in [0-9.]*s//'

........................................................                 [100%]
56 passed

from evidencedesk.contracts import replay

new = [{"name": "six ids", "patch": {"cited_ids": ["a"] * 6}, "expect": "invalid_schema"},
       {"name": "empty answer", "patch": {"answer": ""}, "expect": "invalid_schema"},
       {"name": "version 3", "patch": {"schema_version": 3}, "expect": "unknown_version"}]
for case in new:
    print(f"{case['name']:15} expect={case['expect']:15} got={replay(case)}")

six ids         expect=invalid_schema  got=invalid_schema
empty answer    expect=invalid_schema  got=invalid_schema
version 3       expect=unknown_version got=unknown_version