Chapter 10 · Production Capstone Integration and Release
Capstone Architecture and Integration
- Page 19 of 22
- 25 min read
Nadia in Dhaka has six working pieces: ingestion, a permission-aware retriever, a release gate, a tracer, a safety layer and a streaming service. Her manager asks: "Is it one system?" Each piece passed its own tests. The bugs that hurt are in the joins: a filter built from the wrong identity, a status word one module lacks, a route that came along with a wrapper. (An illustrative case.)
This page opens the capstone chapter. It adds one subpackage, evidencedesk/knowledgeops/, and edits nothing else. It joins the pieces of Ingestion and Permission-Aware Retrieval, Retrieval Operations and Data Freshness, Robust Metrics and Release Gates, Guardrails and Security Regression, Structured Logging and Tracing and Containerisation and Streaming Deployment without teaching them again. The next page collects the release evidence, using the habits of Packaging, Configuration, and Reproducibility and Documentation, Release, and Portfolio Quality. Honest limits: every model here is a fake (real models are less predictable, which is exactly why you test), the clock is fake, the question set has 19 cases, and at the end of this page the release candidate is not ready. That is the correct output, and the page shows why.
What you will learn
- How to define a narrow use case and write machine-checkable acceptance criteria before the code that meets them.
- How to choose between RAG and an agent with a decision record, and how to generate the architecture and trust boundaries from the running system.
- How to make model and tool contracts explicit, and how to keep the provider behind one interface with a fake by default.
- How to wire evaluated retrieval, one offline gate and critical-path instrumentation behind one composition root.
- How to run a full-system offline release candidate, and which integration seams did not fit.
Choose a narrow production use case
A use case is narrow when you can name the user, the task, the data source and the failure boundary: what the system does, in advance, when it cannot answer, when the model is down and when the model's output is malformed. KnowledgeOps is a help assistant for signed-in staff of two companies that share one platform. It answers only from their documents and cites every claim. The definition is data, validated by a Pydantic model, with a problems() check that lists what is too vague to build against.
from evidencedesk.knowledgeops.usecase import load_usecase
uc = load_usecase()
print("task:", uc.task[:95] + "...")
print("out of scope:", "; ".join(uc.out_of_scope[:3]) + "; ...")
print("allowed tools:", uc.allowed_tools, "| tenants:", uc.tenants)
for name, rule in uc.failure_boundary.items():
print(f" {name}: {rule}")
print("problems:", uc.problems())
vague = uc.model_copy(update={"user": " ", "out_of_scope": ["x"], "failure_boundary": {}})
print("vague draft:", len(vague.problems()), "problems, first:", vague.problems()[:2])task: Answer policy and product questions from the company's own documents. Every claim carries a cit...
out of scope: issuing refunds or credits; sending email; answering from the open web; ...
allowed tools: ['search_knowledge', 'read_chunk'] | tenants: ['kestrel', 'harbor']
unsupported_question: abstain with the standard sentence; never guess
forbidden_document: behave exactly as if the document did not exist
provider_down: a plain failure message and a retryable code; no partial answer
malformed_model_output: reject the whole reply; do not repair it
problems: []
vague draft: 6 problems, first: ['user is empty', 'name at least two things that are out of scope']The out-of-scope list is a design tool: "issuing credits" and "sending email" keep this assistant read-only, and the decision below depends on them.
Write acceptance criteria before coding
The file data/knowledgeops/acceptance.json holds 19 criteria in five categories. Each names a measure, a comparison and a number, so a program can decide it. The idea comes from the deployment acceptance file of Containerisation and Streaming Deployment, which names a test per criterion; here most criteria are measurements, so they carry a threshold. check turns a dict of measured values into one verdict per criterion, and a measure nobody produced fails: absence is not a pass.
from collections import Counter
from evidencedesk.knowledgeops import criteria as crit
criteria = crit.load_criteria()
print(dict(Counter(c.category for c in criteria)))
for c in criteria[:2] + criteria[5:6] + criteria[11:12]:
print(f"{c.id} {c.measure} {c.op} {c.value} # {c.requirement[:68]}")
verdicts = crit.check(criteria[:3], {"answer_correct": 0.80, "abstain_correct": 0.95})
print([(v.id, v.ok, v.reason) for v in verdicts]){'functionality': 5, 'security': 6, 'latency': 1, 'observability': 4, 'recovery': 3}
F1 answer_correct >= 0.85 # A signed-in caller gets the stated fact, with a citation, for at lea
F2 abstain_correct >= 0.9 # A question the caller's documents do not answer is abstained or clar
S1 cross_tenant_leaks == 0.0 # No answer or citation carries text from another tenant's document.
L1 p95_fake_s <= 0.5 # p95 request latency on the fixed set stays within 0.5 s of the fake
[('F1', False, 'answer_correct 0.8 is not >= 0.85'), ('F2', True, ''), ('F3', False, 'not measured')]Two honesty rules. Offline, a latency criterion can only be a structural budget on a fake clock, and the file says so. And thresholds come from need and from the earlier baseline (the Beginner assistant passed 0.852 of its fixed cases), not from the first run: moving a number to fit a result is the mistake Robust Metrics and Release Gates warned about. The gate derives its bounds from this file, so the two cannot disagree.
Choose RAG or an agent
Prefer the simpler architecture that meets the requirements. The rule here is mechanical: an agent is chosen only if a need cannot be met without one (a write action, planning with an unknown number of steps, open-ended tool choice). The record reuses DecisionRecord from Fine-Tuning Decisions and LoRA Concepts for its fields and Markdown, with its own problems() because the original accepts only fine-tuning options.
from evidencedesk.knowledgeops.decision import decide_architecture
from evidencedesk.knowledgeops.usecase import Needs
record = decide_architecture(uc)
print(record.to_markdown())
agentic = uc.model_copy(update={"needs": Needs(**{**uc.needs.model_dump(), "write_actions": True})})
r2 = decide_architecture(agentic)
print("if the task also had to issue credits:", r2.decision, r2.triggers)# Decision AR-1 (2026-10-13)
Problem: Answer policy and product questions from the company's own documents. Every claim carries a citation; a question the documents do not answer gets an abstention, not a guess.
Symptom: answers must follow documents that change
Options considered: rag, workflow, agent
Decision: rag
Evidence:
- no requirement needs an agent: write_actions, multi_step_planning, open_ended_tool_choice are all false
- facts change (changing_facts is true), so the documents are retrieved at question time, not trained in
- allowed tools are read-only: search_knowledge, read_chunk
Risks:
- a fixed pipeline cannot handle a question that needs two lookups in sequence
- retrieval quality caps answer quality
- the simpler design has fewer places to hold a permission bug, but the ones it has are in retrieval
Baseline: none
Review when: a requirement adds an action that changes data, or the fixed question set shows multi-step failures
if the task also had to issue credits: agent ['write_actions']An agent would add what Agent Loops, State, and Memory and Human Approval and Multi-Agent Boundaries had to add: a step budget, saved state, approvals. Nothing in this task needs them, and the record says when to revisit.
Draw the data flow and trust boundaries
A hand-drawn diagram drifts. This one is read from the objects the composition root built (shown in full below), and every arrow carries a boundary id from the threat model of Threat Modelling and Injection Testing and the name of the code that enforces it. problems() imports each enforcing function, checks that every boundary is crossed by some flow and that every tenant has a partition.
from evidencedesk.knowledgeops.system import build_system
system = build_system() # the composition root, read in full further down
print(system.ingest_report.text().splitlines()[0])
print(system.ingest_report.counts())run-001 epoch 1 embedded 21 verify findings 0
{'added': 12, 'quarantined': 1}from evidencedesk.knowledgeops import architecture
print(architecture.describe(system))
print("problems:", architecture.problems(system))
print("surfaces the shipped threat model does not list:", architecture.threat_gaps())components
callers untrusted people with a bearer token; 2 tenants share the platform
edge edge stream route: bearer auth, size limit, quota, admission; the inherited /ask route is removed
pipeline core retrieve -> context -> model -> validate; one span per stage; identity from the directory, never from the text
knowledge data 12 sources, epoch 1, 1 quarantined; partitions: harbor 2 chunks, kestrel 19 chunks
model provider LLMClient protocol only (ExtractiveModel offline); no tools, JSON reply contract
tools core read-only contracts: search_knowledge (read), read_chunk (read)
telemetry telemetry events + spans; 13 field names logged as length only; content capture off
flows (boundary of the threat model, enforcing code)
callers --B1--> edge create_stream_app
edge --B1--> pipeline Directory
pipeline --B5--> knowledge security_allowed
pipeline --B2--> model SecretGuard
model --B3--> pipeline check_output
pipeline --B4--> tools ToolBox
pipeline --B7--> telemetry EventLog
problems: []
surfaces the shipped threat model does not list: ['tool:search_knowledge', 'tool:read_chunk', 'input:model_reply', 'route:/v1/stream']Users, API, retrieval, model, tools, storage and logs are all there. The last line is the honest part: the shipped threat model has no entry point for four surfaces this package adds. The next page closes them.
Security note (as of 2026-10-12): The OWASP Top 10 for LLM Applications 2025 edition is the one shown on the project's main page. A 2026 edition dated 2026-08-03 exists, but its list could not be confirmed from an official page. Name the edition you use and check the current documentation.
Define the internal model and tool contracts
Four contracts cross boundaries. AskRequest has no tenant, user or group field, and an unknown field is an error, so a request cannot claim an identity. ModelReply is the only shape the model may return; a tool request or an extra field rejects the whole reply. Answer is what leaves the system. The tools are explicit read-only contracts, and the use case lists which are allowed.
# excerpt of kit/evidencedesk/knowledgeops/contracts.py
class SearchArgs(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True)
query: str = Field(min_length=1, max_length=500)
filters: Filters | None = None # may only narrow what the caller can already read
k: int = Field(3, ge=1, le=5)The tool box validates arguments, checks the allow-list, and runs the permission-aware search of the previous chapter with the caller's identity. It returns a code and never an error message, because Pydantic's message can echo the input.
from evidencedesk.knowledgeops.contracts import boundary_problems
from evidencedesk.knowledgeops.directory import Directory
people = Directory.load()
calls = [("kwame", "search_knowledge", {"query": "goodwill credit limit"}), ("cust-7", "search_knowledge", {"query": "goodwill credit limit"}),
("aiko", "search_knowledge", {"query": "x", "tenant": "harbor"}), ("aiko", "delete_everything", {}),
("cust-7", "read_chunk", {"chunk_id": "kc-payroll-note:c0"}), ("cust-7", "read_chunk", {"chunk_id": "kc-nothing-here:c0"}),
("lena", "read_chunk", {"chunk_id": "kc-payroll-note:c0"})]
for who, tool, args in calls:
out = system.tools.run(people.principal(who), tool, args)
print(f"{who:<7}{tool:<18}{out.code:<18}", [h.source_id for h in out.data.hits] if out.ok and hasattr(out.data, "hits") else "")
print("provider imports in the package:", boundary_problems())kwame search_knowledge ok ['kc-goodwill', 'kc-goodwill', 'kc-fleet-policy']
cust-7 search_knowledge ok ['kc-returns', 'kc-warranty', 'kc-shipping']
aiko search_knowledge invalid_arguments
aiko delete_everything unknown_tool
cust-7 read_chunk not_found
cust-7 read_chunk not_found
lena read_chunk ok
provider imports in the package: []Read the rows: Kwame's search finds the goodwill note that Customer 7's does not; a forbidden chunk and a missing chunk give the same code; Lena (in the HR group) can open what Customer 7 cannot. The provider boundary is checked in code too: boundary_problems() scans every module of the package for provider SDK imports and finds none. A test confirms that the tool path and the pipeline path return the same chunks for the same caller. The RAG pipeline does not call the tool box; the contracts are for an agent extension or a client.
Integrate a provider-neutral LLM layer
llmlayer.py imports no SDK. It builds one request (rules, escaped evidence marked trust="untrusted", the question), defines the reply contract, and ships stand-ins. ExtractiveModel is the default: not a language model, it picks the best-matching sentence of the evidence. ScriptedModel replays given replies or raises a given error, which is how a provider outage and hostile output are rehearsed. A real adapter implements complete(request) and nothing else. Retries use the error classes of Adapters, Configuration, and Error Handling, and their waits run on the fake clock.
Provider note (as of 2026-10-12): One provider's documentation says to check the stop reason first: a refusal comes back as a normal successful response that may not match the schema, and a reply cut off at the token limit may be invalid. The pipeline therefore checks
stop_reasonbefore it parses. Check the current documentation of your provider.
from evidencedesk.knowledgeops import llmlayer, scenarios
block = llmlayer.evidence_block("kc-returns:c0", 1, "Customers may return a new bike within 21 days of delivery.")
request = llmlayer.build_request("How long do I have to return a bike?", [block])
print(request.messages[-1].content)
print(f"{'scenario':<20}{'status':<10}{'code'}")
for sid, (answer, as_expected) in scenarios.run_all().items():
print(f"{sid:<20}{answer.status:<10}{answer.code}{'' if as_expected else ' <-- NOT AS EXPECTED'}")<evidence source="kc-returns:c0" version="1" trust="untrusted">
Customers may return a new bike within 21 days of delivery.
</evidence>
<question source="user-input">
How long do I have to return a bike?
</question>
scenario status code
provider_down failed server
rate_limited failed rate_limited
not_json failed invalid_reply
extra_field failed invalid_reply
truncated failed truncated
no_citation failed unknown_citation
foreign_citation failed unknown_citation
claim_not_in_source abstained unsupported_claim
image_exfil blocked unsafe_output
foreign_link blocked unsafe_output
secret_in_answer blocked unsafe_outputEleven rehearsals end where they must. Look at the order of checks: the output check for links, images and secrets runs before the support check. The first version did it the other way round, and an exfiltration attempt was reported as a harmless "unsupported claim"; safety checks go first, quality checks second.
Integrate evaluated retrieval
The composition root is one function. Everything is injected or defaulted in it, so a test swaps one part and keeps the rest.
# excerpt of kit/evidencedesk/knowledgeops/system.py
def build_system(model: LLMClient | None = None, config: Config | None = None, root: Path = INGEST_DATA, usecase: UseCase | None = None) -> System:
"""The composition root. Everything is injected or defaulted here, so a test can swap one part and keep the rest."""
config, usecase = config or Config(), usecase or load_usecase(DATA / "usecase.json")
clock = ManualClock(1000.0)
guard = SecretGuard({"api_key": SIMULATED_SECRET})
log = EventLog(clock, guard)
kb = KnowledgeBase(root, clock=clock, log=log)
report = kb.ingest(snapshot(config.snapshot))
tracer, ledger = Tracer(clock, CountingIds(), log), Ledger()
base = model or llmlayer.ExtractiveModel(config.min_share, corpus_idf(kb))
retrying = RetryingClient(base, RetryPolicy(max_attempts=config.max_attempts), sleep=clock.advance, rng=random.Random(7))
traced = TracedModel(Relabel(SlowClient(retrying, clock, MODEL_BASE_S, MODEL_PER_TOKEN_S), config.pricing_name), tracer, ledger)
return System(usecase, kb, clock, log, tracer, ledger, traced, guard, ToolBox(kb, usecase.allowed_tools), config, report)Retrieval is not rebuilt. KnowledgeBase from Ingestion and Permission-Aware Retrieval keeps one hybrid index per tenant and filters by identity before ranking; fallback.decide from Retrieval Operations and Data Freshness answers, asks, caveats or abstains. The pipeline adds spans, the model call and validation around them. The package is thirteen small modules: definitions (usecase, criteria, decision), contracts (contracts, tools, directory), the model layer (llmlayer, scenarios), the root (system) and the outer parts (gate, architecture, app, rc).
Q = "How long do I have to return a bike?"
for name in ("cust-7", "omar", "aiko"):
a = system.ask(people.principal(name), Q, f"demo-{name}")
print(f"{name:<7}{a.status:<10}{a.code:<6}{a.text}")
a = system.ask(people.principal("aiko"), "How much goodwill credit can an agent give?", "demo-restricted")
print("aiko, goodwill:", a.status, a.code, "|", a.text[:52])cust-7 answered ok Customers may return a new bike within 21 days of delivery. [kc-returns:c0]
omar answered ok Harbor Bikes accepts returns within 30 days of delivery. [hb-returns:c0]
aiko answered ok Customers may return a new bike within 21 days of delivery. [kc-returns:c0]
aiko, goodwill: abstained low_coverage | Not enough evidence in the help centre to answer thiCustomer 7 and Aiko share a tenant and get the 21-day window; Omar's token maps to the other tenant and gets 30. Aiko asks about the goodwill limit, which only team leads may read, and receives the standard abstention: the document behaves as if it did not exist. Now the one number the acceptance criteria will test, the model's abstention threshold:
from evidencedesk.knowledgeops.gate import answer_report
from evidencedesk.knowledgeops.system import Config
print("min_share answer_correct abstain_correct wrong answers on unanswerable cases")
for share in (0.3, 0.4, 0.5, 0.6):
r = answer_report(build_system(config=Config(min_share=share)))
c = r.counts
print(f"{share:<11}{c['answer_correct'].k}/{c['answer_correct'].n:<14}{c['abstain_correct'].k}/{c['abstain_correct'].n:<15}{c['abstain_correct'].n - c['abstain_correct'].k}")min_share answer_correct abstain_correct wrong answers on unanswerable cases
0.3 9/12 6/7 1
0.4 8/12 7/7 0
0.5 8/12 7/7 0
0.6 6/12 7/7 0No setting wins. A permissive model answers more but gives one wrong answer to a question the documents do not cover; a strict one abstains correctly and answers fewer. A wrong answer costs more than an abstention here, so the default is 0.5. The counts are 12 and 7 cases, so a difference of one case is within noise; the table is evidence for a choice, not a ranking.
Add an offline evaluation gate
One command runs three kinds of evidence and prints one verdict, the worst of the parts. Each part is an evalkit report judged by the gate of Robust Metrics and Release Gates: the end-to-end answers (fixed cases, citation and leak checks), the retrieval regression suite of Retrieval Operations and Data Freshness, and the attack corpus of Guardrails and Security Regression. The answers part takes its bounds from the acceptance file and its critical cases from the cases file. Only a block makes the exit code 1.
from evidencedesk.knowledgeops.gate import run_gate
result = run_gate(system)
print(result.text())
print("exit code:", result.exit_code())part verdict reasons that are not notes
answers block floor: answer_correct 0.667 (8/12) is below the floor 0.85; floor: citation_support 0.8 (8/10) is below the floor 0.9
retrieval go -
security go -
GATE: BLOCK
exit code: 1The gate blocks on answer quality and citation support, and the other two parts are clean. Both numbers have a cause you can read: paraphrases ("three weeks" for 21 days, "broken" for damaged) and a table row that the structure-aware chunker split from its header. Two more facts matter. The retrieval part needs the right snapshot: the older one blocks with a stale return window. And the answers part is compared with a stored baseline, a record of the first run and not an approval.
Can the gate see a permission bug? Here the filter is deliberately built from the wrong identity, in a temporary patch, and the cases are re-run.
from unittest import mock
from evidencedesk.ingest.contract import Principal
from evidencedesk.ingest.fallback import decide as real_decide
from evidencedesk.knowledgeops import system as system_module
from evidencedesk.knowledgeops.criteria import load_criteria
from evidencedesk.knowledgeops.gate import answer_report, evaluate, load_cases, policy_from_criteria
boss = lambda p: Principal(p.user_id, "kestrel", ("staff", "team-leads", "hr")) # the bug: the filter is built from the wrong identity
with mock.patch.object(system_module, "decide", lambda kb, p, q, *a, **k: real_decide(kb, boss(p), q, *a, **k)):
bad = answer_report(build_system())
decision = evaluate(policy_from_criteria(load_criteria(), bad, load_cases()), bad)
print(decision.verdict, {k: bad.scalars[k] for k in ("cross_tenant_leaks", "restricted_leaks")})
print([r.text for r in decision.reasons if r.rule == "critical"][:3], "...")block {'cross_tenant_leaks': 2.0, 'restricted_leaks': 4.0}
['case k12 failed', 'case k13 failed', 'case k14 failed'] ...The pipeline reported nothing; it believes its own filter. The gate found the bug with an independent scan: leaks looks for sentences of chunks the caller may not read inside the answer, and ignores a sentence repeated in a public page (the first version flagged "Clearance items and gift cards cannot be returned", which is in both). Critical cases make one leak block, whatever the averages say.
Instrument the critical path
Every request leaves one trace: a root span and one span per stage, with safe attributes only. User ids are pseudonyms, content capture is off (lengths are kept), and a span carries a stable code, never a message. The same data feeds the usage ledger and the observations of Monitoring, Budgets, and Incidents.
from collections import Counter
from evidencedesk.observe import viewer
print(viewer.tree(system.tracer.traces()[0]))
print("codes:", dict(sorted(Counter(o.code for o in system.observations).items())))
t = system.ledger.summary()
print(f"model calls {t.calls}, tokens {t.input_tokens}+{t.output_tokens}, invented cost {t.cost}")
import json
print(json.dumps(json.loads(system.log.lines[-1])["fields"]), json.loads(system.log.lines[-1])["event"])request +0.000s 0.296s code=ok model_calls=1 outcome=answered status=answered tenant=kestrel
├─ retrieval +0.000s 0.030s action=answer coverage=0.5 hits=3 reason=ok top_score=0.0328
├─ context +0.030s 0.004s evidence=3 flagged=0
├─ model +0.034s 0.260s cost=0.000852 model=fake-model tokens_in=164 tokens_out=30 usage_estimated=False
└─ validation +0.294s 0.002s code=ok
codes: {'clarify': 1, 'low_coverage': 6, 'model_abstained': 3, 'ok': 13}
model calls 16, tokens 2503+429, invented cost 0.012657
{"hits": 3, "k": 3, "late_denied": 0, "query_chars": 28} search.queryThe error categories are the code column: retrieval abstentions (low_coverage) never reach the model and cost nothing, while model_abstained costs tokens. Latency is the fake clock: 0.296 s here is stage constants plus a pretend model, so it proves the structure, not the speed. A real latency number needs a real model and a load test (Scaling, Resilience, and Rollback).
Exercise: a full-system offline release candidate
One command runs every step with fixtures and fakes: the definitions, the build, the architecture check, the gate, the HTTP app, the failure rehearsals and the telemetry checks. It measures each acceptance criterion and prints a compact table. It reports; deciding go or no-go is a person's job, with this evidence.
python -m evidencedesk.knowledgeops --criteria
echo "exit code: $?"step ok detail
use case yes Kestrel staff help assistant: 5 out-of-scope items, 2 tenants
decision yes rag over workflow, agent
criteria yes 19 criteria in 5 categories, all measurable
ingest yes {'added': 12, 'quarantined': 1}; verify findings 0
contracts yes 2 tool contracts, provider imports in package: 0
architecture yes 7 flows, threat-model gaps to close later: 4
gate NO answers=block retrieval=go security=go -> block
app yes routes /health, /ready, /v1/stream; 5 streams; no token -> 401
failures yes 11/11 scenarios ended where they must; stale snapshot blocked: True
telemetry yes 24 traces, 192 log lines, 18 model calls, 3362 tokens
id measure value result
F1 answer_correct 0.667 FAIL
F2 abstain_correct 1.0 ok
F3 citation_resolves 1.0 ok
F4 citation_support 0.8 FAIL
F5 stream_complete 1.0 ok
S1 cross_tenant_leaks 0.0 ok
S2 restricted_leaks 0.0 ok
S3 unsafe_outputs_served 0.0 ok
S4 attacks_stopped 0.97 ok
S5 unauthenticated_http 401.0 ok
S6 legacy_route_open 0.0 ok
L1 p95_fake_s 0.31 ok
O1 traces_well_formed 1.0 ok
O2 error_codes_present 1.0 ok
O3 usage_per_model_call 1.0 ok
O4 log_content_leaks 0.0 ok
R1 provider_failure_clean 1.0 ok
R2 malformed_reply_rejected 1.0 ok
R3 bad_snapshot_blocked 1.0 ok
criteria passing 17/19; failing: F1, F4
RESULT: NOT READY
exit code: 1Ten steps, 17 of 19 criteria met, one verdict. The two failures are the ones the gate named; everything that guards safety passed: no leak, no unsafe output served, 401 without a token, the inherited route gone, a clean provider failure, a malformed reply rejected whole, a stale snapshot blocked. "Not ready" for quality with "ready" for safety is a usable answer; one average would have hidden it. The HTTP seam, through the streaming service of Containerisation and Streaming Deployment, wrapped and unedited:
from fastapi.testclient import TestClient
from evidencedesk.knowledgeops import app as ops_app
with TestClient(ops_app.create_ops_app(build_system())) as http:
print(ops_app.route_paths(http.app), http.get("/ready").json())
for name, q in (("cust-7", "How long do I have to return a bike?"), ("omar", "How long do I have to return a bike?"),
("cust-7", "When are salary reviews held?"), ("", "hello")):
code, kinds = ops_app.smoke(http.app, name, q)
print(f"{name or '(no token)':<11}HTTP {code} {' '.join(kinds)}")['/health', '/ready', '/v1/stream'] {'ready': True, 'reason': '', 'chunks': 22, 'offline': True}
cust-7 HTTP 200 start token token token token token done
omar HTTP 200 start token token token token done
cust-7 HTTP 200 start token token token token token token done
(no token) HTTP 401 The bearer token's name decides the identity (the body cannot carry one), the stream ends with done, a restricted question gets the abstention, and no token gets 401 before any work. The tests for every seam:
python -m pytest tests/test_knowledgeops_definition.py tests/test_knowledgeops_contracts.py tests/test_knowledgeops_system.py \
tests/test_knowledgeops_gate.py tests/test_knowledgeops_app.py tests/test_knowledgeops_rc.py 2>&1 | tail -1 | sed 's/ in [0-9.]*s//'86 passedSeams that did not fit, and what is not integrated
| Where | Mismatch | What I did |
|---|---|---|
DecisionRecord | accepts only fine-tuning options | subclass with its own problems() |
create_stream_app | brings the Beginner /ask and wants a retriever for readiness | a length-only view for /ready; the route is removed from the router after the call |
Upstream.start | receives a token name, not a tenant or groups | a directory maps names to principals |
observe metrics | statuses answered/refused/blocked/failed only | abstained and clarify map to refused; the full outcome stays in a span attribute |
threat.egress.check_output | expects the Beginner reply shape | re-wrap the answer before the check |
ingest.access.Hit | snippet only, 160 characters | read the full chunk from the knowledge base (a reach into its records) |
observe.pipeline.Desk, price table | built for the Beginner retriever; fake model names only | spans written again (duplication); Relabel |
evalkit.ci suite | its baseline is the Beginner corpus and it blocks by its own baseline | not part of this gate |
No dependency was added to the requirements files: the package needs only what the kit installs, and anyio comes with FastAPI.
Not integrated: the MCP server (the tool box is the contract; the wire is not used), the agent loop, reranking (the strict-model misses are where it would be measured), the retrieval cache, trace export, a judge, a real provider adapter and a real container run. Streaming cuts pieces from a finished answer; it is not token streaming from a model.
Where this shows up at work
A first review asks "what is it for, and what happens when it cannot answer" long before it asks about models. The use case, the acceptance file, the decision record and the one gate command are what you hand a reviewer, and what lets a second engineer change one part and see within a minute whether the whole holds.
Common mistakes
- Criteria written after the first run. The thresholds then describe the system. Write them from need, record the baseline, and fail a missing measure.
- Taking identity from the request. Tenant and groups come from the login, mapped in code. A body field with the same name is an attack surface.
- Trusting the filter to report its own bugs. Check answers with a scan that does not use the filter, and mark leak cases critical.
- Quality checks before safety checks. A leaking answer then shows up as "unsupported". Run the output check first.
- One average for release. Report each part and each criterion; a strong average hides a failed one.
Try it yourself
- Easy (revisits Rubrics, Regression, and Release Gates): build the system from the older snapshot and run the gate. Predict which part blocks for a reason that has nothing to do with the model.
- Medium (revisits Budgets, Caching, and Usage Limits): group the requests served so far by error code and compare the mean invented cost. Predict which abstentions are free.
- Hard (the exercise, applied): run the release candidate with abstention thresholds 0.3 and 0.5 and compare the failing criteria. Predict whether the permissive model passes more or fewer, then say which setting you would ship and why.
Answers
Easy. The retrieval part: the older snapshot still has the Bangla returns page at 14 days, so the stale-value check and a critical case fail, whatever the model does.
from evidencedesk.knowledgeops.system import Config
stale = run_gate(build_system(config=Config(snapshot="v2")))
print({p.name: p.decision.verdict for p in stale.parts})
print([r.text for r in stale.parts[1].decision.reasons if r.level == "block"]){'answers': 'block', 'retrieval': 'block', 'security': 'go'}
['case r16 failed', 'evidence_at_3 0.812 (13/16) is below the floor 0.9', 'stale_free 0.667 (2/3) is below the floor 1.0']Pass checklist: retrieval is block, security is go, and the reasons name a stale value, not a model failure.
Medium. Retrieval abstentions never call the model, so they cost nothing; a model abstention cost tokens, about three quarters of an answered request.
from collections import defaultdict
cost = defaultdict(list)
for o in system.observations:
cost[o.code].append(o.cost)
for code, rows in sorted(cost.items()):
print(f"{code:<16}requests {len(rows):<3}mean invented cost {sum(rows) / len(rows):.6f}")clarify requests 1 mean invented cost 0.000000
low_coverage requests 6 mean invented cost 0.000000
model_abstained requests 3 mean invented cost 0.000640
ok requests 13 mean invented cost 0.000826Pass checklist: low_coverage and clarify are exactly zero; model_abstained is not; you grouped by code, not by status (both are "refused" in the metrics).
Hard. The permissive model fails more criteria: F2 joins F1 and F4 because it answers a question the documents do not cover.
from evidencedesk.knowledgeops import rc
from evidencedesk.knowledgeops.system import Config
for share in (0.3, 0.5):
steps, verdicts, gate = rc.run_rc(Config(min_share=share))
print(share, "failing:", [v.id for v in verdicts if not v.ok], "| gate", gate.verdict)0.3 failing: ['F1', 'F2', 'F4'] | gate block
0.5 failing: ['F1', 'F4'] | gate blockPass checklist: you predicted before running; you name F2 as the new failure; you ship 0.5, because a wrong answer to an unanswerable question is worse than an extra abstention, and you note that neither setting meets F1, so the real fix is better retrieval or a stronger model, not a threshold.
Summary
- Define the use case, the failure boundary and the acceptance criteria as data before the code; a missing measure fails.
- Choose RAG unless a need requires an agent, in a decision record that says when to revisit.
- Generate the architecture and its trust boundaries from the running system, and list the surfaces the threat model misses.
- Keep identity out of requests, the provider behind one interface with a fake default, and safety checks before quality checks.
- One composition root, one gate over answers, retrieval and security, and one trace per request; run it all offline and read the table, including the failures.
Next: Release Evidence and Production Readiness. The next page stress-tests this release candidate, records its versions and limits, writes the threat model, runbook and rollback plan, and ends with an explicit go or no-go.