Chapter 3 · Streaming, Structured Outputs, and Reliable Responses
Reliable Output Contracts
- Page 6 of 22
- 21 min read
Kwame in Accra ships the answer box from the last page. Then the team adds a field and an old mobile app crashes, a stream dies half way and the screen shows half a sentence as final, and a model returns a link that points somewhere else. These are contract problems: nobody wrote down what the front end can rely on. This page reuses AnswerResult, Outcome, assemble and FakeStreamClient from Streaming and Structured Results and turns them into a contract.
What you will learn
- Schema versions, and one predictable result for empty, partial and error states.
- Meaningful stream events, cancellation, Unicode chunk boundaries and size bounds.
- Returned ids and URLs as untrusted data, provider contracts, regression fixtures, and a tested streaming FAQ endpoint.
Use schema versioning
A schema version is a field, schema_version, that says which shape follows, so a change is a decision and not a surprise. Version 2 adds links. The version-aware reader read_answer accepts 1 and 2, turns both into one internal model (Answer), and rejects every other value, a missing one included, instead of guessing. Version 1 keeps its old rules, so a v1 reply with links is invalid. (type(version) is int because True == 1 in Python.) The module is shown in parts, in file order; the blocks after the first continue the same file (the kit has it whole). parse_contract is the entry point: stop reason, then size, then JSON syntax, then version and schema.
# 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=-A new required field, a rename or a changed meaning needs a new version; keep a reader while old clients or data exist.
Define empty, partial and error states
Without a rule every screen invents its own. The Envelope is the one shape the front end always receives, with a state of ok, empty, partial or error; to_envelope builds it from an Outcome (structured answer) or an Assembled stream (plain text).
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'The front-end rule: always render text; use state for styling (a "cut off" label on partial), retryable for a retry button, result only when set. Errors show a fixed message; the code is for logs. Partial exists only for plain text, because half of a JSON text is not an answer. Empty is not an error: the call worked and said nothing. A dropped stream may work next time; a reply cut by the length limit will be cut again.
Keep streaming events meaningful
An event is a content chunk, a completion (done) or an error; the last two are terminal. The contract: known kinds, no empty chunk, exactly one terminal event, nothing after it. check_events returns the broken rules; run it on what the server sends and the client receives.
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']An interrupted stream has no terminal event; a clean error event is terminal. A valid sequence says nothing about whether the answer is right.
Handle cancellation and client disconnects
When a user closes the tab, the server may keep reading the model's stream. The principle: stop reading, close the stream, and let the SDK documentation tell you what stops on the provider's side (VERIFY). Your part is testable: a generator closed early runs its finally. watched logs what its source produced and whether it was closed; bounded (the stream size limit) closes its source when it stops.
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']After two reads the source had produced two events and close() ran the finally. This shows that our code releases its source, not that a provider stops generating or billing.
Provider note (as of 2026-10-12): FastAPI documents that a task can be cancelled only at an
await(a generator without one "can not be cancelled properly") and that its "Stream Data" page handles cancellation for you. Starlette documentsawait request.is_disconnected()for streaming responses; itsTestClientdoes not emulate concurrent request and response streaming, so it cannot test a real disconnect (use a running server). Anthropic documents that leaving thewithblock closes the response, not what the server still bills; Gemini's cancel call is for background interactions, not a foreground stream; for OpenAI and Ollama the pages checked do not document cancelling a stream early. VERIFY each. Check the current documentation.
Test Unicode chunk boundaries
Reassembly must be lossless: join the chunks and compare. Display is what a screen draws if it shows one chunk at a time. Bangla, Hindi and accented Latin build one visible unit, a grapheme cluster, from several code points: কি is a consonant plus a vowel sign, ক্ষ joins two consonants with a virama, é can be e plus a combining accent. The standard library has no full cluster splitter (UAX #29 needs a library), so clusters is an approximation; broken_boundaries counts boundaries inside a cluster.
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']Splitting by characters is lossless every time (the pieces are slices of one string), yet in the Bangla text it leaves 10 boundaries inside a cluster; the second piece is BENGALI VOWEL SIGN I alone. A page that draws each chunk in its own element shows a vowel sign with nothing to attach to. We measure the boundary, not the pixels. Cluster-aware chunks give 0. So the front end appends chunks to one string and renders that; tests try many chunk sizes, because a boundary can be harmless by luck (size 3 on the accent text); and text is compared after unicodedata.normalize, because é as one code point or two differs for ==.
Bound output size
Three limits: one reply (MAX_BYTES, checked before parsing), one stream (bounded), and schema fields (answer up to 1000 characters as before; at most 5 ids and 5 links; ids of a safe pattern; links up to 200 characters). Over a limit you can reject or truncate and flag. We reject: a cut answer can lose the "not" or the warning at its end, a cut list drops citations, and cut JSON is invalid anyway. A rejected reply becomes a safe error with a code you can count (too_large, also in the fixtures below). Truncate-and-flag fits only plain text with a visible cut.
Validate URLs and identifiers as data
A schema checks shape, not meaning. A model can return a well-formed id it never saw, or a valid URL to another site; retrieved text can steer it there (see Context Assembly and Trust Boundaries). Two checks, pure text, no network call: cited ids must belong to the set retrieved for this request (stricter than "exists in the corpus"), and each link must pass check_url: scheme https, no credentials, host exactly on an allowlist. Use urlsplit: startswith is fooled by help.kestrelcycles.example.evil.test and help.kestrelcycles.example@evil.test.
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 httpsEach trick fails for its own reason; the last two lines are the report you would log. An allowed link does not prove the page supports the claim, and the server should never fetch a model-provided URL.
Compare provider output contracts
Providers differ in where the text is, how a cut-off or refusal is signalled, and what usage fields are called. An adapter maps one shape to our LLMResponse; one parser and one schema then serve all. The two dicts below are hand-written and simplified, not real responses. An unknown signal becomes other, which parse_contract rejects: the safe default.
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_stopProvider note (as of 2026-10-12): Anthropic's structured-output subset does not support
minLength/maxLength, so size limits stay yours. Check before parsing: OpenAI responses can hold arefusalcontent part (do not parse it as JSON) orstatus == "incomplete"withincomplete_details.reason == "max_output_tokens"; Anthropicstop_reasoncan berefusal(a 200 response that may not match the schema) ormax_tokens; GeminifinishReasonhas values other thanSTOP, not to be parsed blindly; Ollama hasdone_reason(values unconfirmed). VERIFY the fields for your SDK version. Check the current documentation.
Write regression fixtures for failures
Every bad reply you meet is worth a line in a file. A regression fixture keeps the reply (raw text, or a patch on a known-good reply), its stop reason and the code you expect, so a change cannot quietly reopen the failure. Add the line first, watch it fail, then fix the code. JSON Lines, one object per line:
{"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"}replay returns the code for one line, and the kit test makes each line one pytest case (the Hard exercise below replays new lines). The file is a floor, not a proof: it holds only failures you have seen.
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)).codeMini-lab: Streaming Structured FAQ Endpoint
Now an endpoint, extending the make_app pattern. Half of a JSON text is not valid, so a structured answer cannot be shown while it streams. The server therefore collects the stream (bounded), validates it (version, schema, ids, links) and only then sends the validated text as chunks and exactly one terminal event carrying the Envelope. The cost is a later first chunk. The status is always 200 (it is sent before any problem is known), so a failure is a single error event whose envelope has the state and 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 appThe fake model is a stand-in (real models are less predictable, which is exactly why you test). Five situations:
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 okGood and Bangla answers stream three chunks and end with done. Every failure sends no chunk at all (no half answer, no unverified link or id) and one error event with its code; check_events finds no broken rule. tests/test_contracts.py asserts this plus the version, URL, adapter, Unicode and fixture cases:
python -m pytest -q tests/test_contracts.py | sed 's/ in [0-9.]*s//'........................................................ [100%]
56 passedThat is the pass line; sed removed the seconds. It does not show how a real stream splits text, whether an SDK cancels, or what a browser draws.
Where this shows up at work
A front-end colleague asks for "a stable response", a product manager asks what users see on failure, a security reviewer asks where links come from. Envelopes, allowlists and fixtures answer all three, and let Lena in Berlin replay yesterday's incidents before approving a change.
Common mistakes
- Changing a field without a version. Old clients break silently. Bump the version, keep the old reader.
- Showing partial structured output. Cut JSON is not an answer. Show partial text only for plain text, marked.
- Several terminal events, or none. The client waits forever or shows two results. Check sequences in tests.
- Testing Unicode with one chunk size. Boundaries can be lucky. Try many sizes, with Bangla.
- Trusting a valid id or URL. A passing schema says nothing about where a value points. Use allowlists, and reject what exceeds a bound instead of cutting it silently.
Try it yourself
- Easy. Predict, then run
check_urlonhttps://help.kestrelcycles.example:8443/x,HTTPS://help.kestrelcycles.example/andhttps://help.kestrelcycles.example./x. Which are allowed? - Medium. Revisit Streaming and Structured Results: stream the delivery question with
stop_after=1and withfail_after=1. What docheck_eventsandto_envelope(assemble(...))say? Why do they differ? - Hard. Write three fixture cases (six ids; an empty answer; version 3), predict their codes, replay them.
Answers
Easy. The first two are allowed: the port is not part of the host, and urlsplit lowercases scheme and host. The third prints host 'help.kestrelcycles.example.' is not allowed: the trailing dot makes it a different string, and refusing is the safe default.
Medium. The dropped stream has no terminal event, so check_events complains and the envelope is partial. The failed stream ends with an error event, a legal terminal event, so the sequence is valid while the envelope is error. One check asks "is the stream well formed?", the other "what may the user see?".
Hard. Six ids break the list limit, an empty answer breaks the minimum length from the last page, and version 3 is unknown. Lines like these, appended to the fixture file, are replayed by the kit test unchanged.
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_versionPass checklist:
- Easy: allowed, allowed, refused, and you can say why (run it to confirm).
- Medium:
stop_aftergives a complaint andpartial;fail_aftergives a valid sequence anderror. - Hard: each got-code equals your expected code, written before replaying.
- Lab: the tests pass and each failure is a one-event stream with a fixed message.
Summary
- Version the schema, read known versions through one reader, reject the rest; give the front end one
Envelopefor every state. - A stream has chunks and exactly one terminal event. For cancellation, stop reading and close the source; the provider side is VERIFY.
- Reassembly is lossless by characters, but display needs whole clusters.
- Reject what exceeds the bounds. Returned ids and URLs are untrusted: allowlists, no network call.
- Adapters map provider shapes into one schema; fixtures keep failures.
Next: Embeddings and Similarity starts the retrieval half of EvidenceDesk: it turns text into vectors, so you can measure how close a question is to a passage.