ক্যাপস্টোন আর্কিটেকচার ও ইন্টিগ্রেশন

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']

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')]

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']

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']

# 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)

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: []

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_output

# 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)

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 thi

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              0

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: 1

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'] ...

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.query

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: 1

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  

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 passed

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']

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.000826

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 block