ইনজেশন ও অনুমতি-সচেতন রিট্রিভাল

# excerpt of kit/evidencedesk/ingest/contract.py
class Source(BaseModel):
    """One entry of a snapshot: the file and everything the file itself cannot say. No default for tenant, owner or licence."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    source_id: str = Field(pattern=r"^[a-z0-9][a-z0-9-]{2,39}$")
    path: str = Field(min_length=1)
    tenant: str = Field(min_length=1)
    owner: str = Field(min_length=1)
    allow: tuple[str, ...]
    sensitivity: Literal["public", "internal", "restricted"]
    licence: str = Field(min_length=1)
    origin: str = Field(min_length=1)
    lang: Literal["en", "bn"] = "en"

from evidencedesk.ingest.lab import snapshot

for e in snapshot("v1"):
    print(f"{e['source_id']:<16}{e['path'].rsplit('.', 1)[1]:<5}{e['tenant']:<8}{e['sensitivity']:<11}{e['allow'][0]}")

kc-returns      md   kestrel public     tenant:kestrel
kc-returns-faq  md   kestrel public     tenant:kestrel
kc-returns-bn   md   kestrel public     tenant:kestrel
kc-shipping     html kestrel public     tenant:kestrel
kc-warranty     csv  kestrel public     tenant:kestrel
kc-stores       json kestrel public     tenant:kestrel
kc-fleet-policy txt  kestrel internal   group:staff
kc-goodwill     md   kestrel internal   group:staff
kc-payroll-note md   kestrel restricted group:hr
hb-returns      md   harbor  public     tenant:harbor
hb-recall       md   harbor  internal   group:staff
kc-promo        html kestrel public     tenant:kestrel

discover   entry valid? format accepted? file present and small enough?
parse      strict UTF-8, no NUL bytes; every format -> title + '## ' sections + '|' tables
clean      NFC, no invisible characters; SCREEN hidden text, injection words, secret and personal-data shapes
classify   tenant known? ACL well formed? label not looser than the content suggests?
chunk      structure-aware chunks (tables keep their header, code is never cut)
dedupe     identical text is indexed once inside one tenant AND one ACL
index      one partition per tenant: BM25 + vectors; each chunk carries its ACL
verify     do manifests, chunk records and index copies still agree?

from evidencedesk.ingest.contract import DATA
from evidencedesk.ingest.parse import fix_pdf_breaks, parse

raw = (DATA / "files/kc-fleet-policy.txt").read_text(encoding="utf-8")
print("\n".join(raw.splitlines()[:5]))
print("-- rebuilt --")
print(fix_pdf_breaks(raw).split("\n\n")[:2])
page = parse(".html", (DATA / "files/kc-shipping.html").read_bytes(), "x")
print("-- html --", page.title, page.findings, "|", page.text.splitlines()[0], "|", page.text.splitlines()[4])
print("-- hostile html --", parse(".html", (DATA / "files/kc-promo.html").read_bytes(), "x").findings)

KESTREL CYCLES - INTERNAL
1. Demo bike lending
Staff may borrow a demo bike for up to three days. The borrower signs the bike out and
returns it with a full char-
ge, clean, with all accessories.
-- rebuilt --
['## Demo bike lending', 'Staff may borrow a demo bike for up to three days. The borrower signs the bike out and returns it with a full charge, clean, with all accessories.']
-- html -- Shipping and delivery () | ## Delivery times | | Speed | Days | Cost |
-- hostile html -- ('hidden_text',)

# excerpt of kit/evidencedesk/ingest/contract.py
    def permits(self, who: Principal) -> bool:
        return self.tenant == who.tenant and bool(self.allow & who.tokens)

    def tokens(self) -> frozenset[str]:
        return frozenset({f"user:{self.user_id}", f"tenant:{self.tenant}", *(f"group:{g}" for g in self.groups)})

# excerpt of kit/evidencedesk/ingest/kb.py
        cur = self.manifests.get(sid)
        cur = cur if cur and cur.status == "active" else None
        meta = dict(owner=src.owner, licence=src.licence, origin=src.origin, lang=src.lang, title=title, fmt=fmt)
        if cur and cur.content_hash == digest:  # same content: at most the permissions or the metadata changed
            same_meta = all(getattr(cur, k) == v for k, v in meta.items())
            if cur.acl == acl and same_meta:
                return Outcome(sid, "unchanged", cur.version, len(cur.chunk_ids))
            return self._swap(cur, replace(cur, acl=acl, ingested_at=now, run_id=run_id, **meta), "acl_updated" if cur.acl != acl else "meta_updated")

from evidencedesk.ingest.lab import fresh, snapshot

kb, report = fresh("v1")
print(report.text())
again = kb.ingest(snapshot("v1"))
print("second run:", again.counts(), "| embedded", again.embedded, "| epoch", again.epoch, "| history rows", len(kb.history))

run-001  epoch 1  embedded 18  verify findings 0
  hb-recall        added        v1 chunks 1
  hb-returns       added        v1 chunks 1
  kc-fleet-policy  added        v1 chunks 3
  kc-goodwill      added        v1 chunks 2
  kc-payroll-note  added        v1 chunks 1
  kc-promo         quarantined  v0 chunks 0  [clean] hidden_text
  kc-returns       added        v1 chunks 2
  kc-returns-bn    added        v1 chunks 1
  kc-returns-faq   added        v1 chunks 2
  kc-shipping      added        v1 chunks 3
  kc-stores        added        v1 chunks 1
  kc-warranty      added        v1 chunks 2
second run: {'quarantined': 1, 'unchanged': 11} | embedded 0 | epoch 1 | history rows 12

manifest = kb.manifests["kc-returns"].as_dict()
for key in ("source_id", "version", "status", "content_hash", "tenant", "owner", "acl", "licence", "origin", "ingested_at", "run_id", "chunk_ids", "stages"):
    print(f"{key:<13}{manifest[key]}")

source_id    kc-returns
version      1
status       active
content_hash 7f871158317e
tenant       kestrel
owner        returns-team
acl          {'tenant': 'kestrel', 'allow': ['tenant:kestrel'], 'sensitivity': 'public'}
licence      synthetic-course-data
origin       help-centre-export
ingested_at  1000.0
run_id       run-001
chunk_ids    ['kc-returns:c0', 'kc-returns:c1']
stages       ['discover', 'parse', 'clean', 'classify', 'chunk', 'dedupe', 'index']

v2 = kb.ingest(snapshot("v2"), prune=True)
print(v2.text().splitlines()[0])
for o in v2.outcomes:
    if o.outcome != "unchanged":
        print(f"  {o.source_id:<17}{o.outcome:<12}v{o.version} {'; '.join(o.reasons)}")
print([(m.version, m.status, m.run_id) for m in kb.history if m.source_id == "kc-returns"])

run-003  epoch 3  embedded 2  verify findings 0
  kc-returns       updated     v2 
  kc-goodwill      acl_updated v1 
  kc-promo         quarantined v0 hidden_text
  kc-recall-notice added       v1 
  kc-stores        deleted     v1 removed_from_source
[(1, 'active', 'run-001'), (1, 'superseded', 'run-001'), (2, 'active', 'run-003')]

from evidencedesk.ingest.access import search
from evidencedesk.ingest.lab import CUSTOMER

rec = kb.records["kc-returns:c0"]
print(rec.chunk.id, "| source", rec.source_id, "v%d" % rec.version, "| ingested in", rec.run_id, "| chunk hash", rec.chunk_hash, "| acl", sorted(rec.acl.allow))
print("citation:", search(kb, CUSTOMER, "How long do I have to return a bike?", 1).hits[0].citation)

kc-returns:c0 | source kc-returns v2 | ingested in run-003 | chunk hash 7235da72 | acl ['tenant:kestrel']
citation: kc-returns v2, Return window, lines 2-3

import tempfile
from pathlib import Path

from evidencedesk.ingest import bait
from evidencedesk.ingest.kb import KnowledgeBase

with tempfile.TemporaryDirectory() as folder:
    messy = KnowledgeBase(root=Path(folder))
    result = messy.ingest(bait.write_messy(Path(folder)))
for o in sorted(result.outcomes, key=lambda o: (o.stage, o.source_id)):
    print(f"{o.stage or 'ok':<9}{o.source_id:<13}{'; '.join(o.reasons)}")

ok       tidy-me      
classify mislabelled  sensitivity_mismatch:restricted
classify no-acl       acl_empty
classify other-tenant unknown_tenant; acl_too_broad
classify wide-open    acl_too_broad
clean    blank        empty
clean    has-contact  personal_data:email
clean    has-key      secret_shaped
clean    shouting     injection_signals:override,secret_request
discover gone         missing_file
discover huge         too_large
discover no-tenant    bad_metadata:tenant
discover tool-exe     unsupported_format
parse    bad-json     bad_json
parse    latin1       bad_encoding
parse    nul-bytes    binary
parse    ragged       ragged_table

from collections import Counter

from evidencedesk.ingest.access import CachedSearch
from evidencedesk.ingest.lab import LEAD
from evidencedesk.ingest.verify import erasure_residue, remember

cache = CachedSearch(kb)
cache.ask(LEAD, "goodwill credit limit")  # an answer that quotes the source is now in the response cache
saved = remember(kb, "kc-goodwill")
print("before:", dict(sorted(Counter(r.split(":")[0] for r in erasure_residue(kb, saved)).items())))
kb.delete("kc-goodwill", "erasure request 12")
print("after: ", erasure_residue(kb, saved), "| tombstone:", kb.manifests["kc-goodwill"].status, kb.manifests["kc-goodwill"].reasons)

before: {'cache0': 1, 'history': 3, 'index': 2, 'manifest': 1, 'record': 2, 'vector': 2}
after:  [] | tombstone: deleted ('erasure request 12',)

# excerpt of kit/evidencedesk/ingest/access.py
def rank(kb: KnowledgeBase, who: Principal, query: str, k: int = 3, filters: Filters | None = None) -> tuple[RankedList, int]:
    """(candidates, late denials). The one place where retrieval and permissions meet."""
    allowed = narrow(kb, security_allowed(kb, who), filters)  # 1. filter first
    part = kb.parts.get(who.tenant)
    found = part.hybrid.search(query, k, allowed) if allowed and part and part.hybrid else []  # 2. rank only what is allowed
    kept = [c for c in found if (m := kb.manifests.get(c.chunk.doc_id)) and m.status == "active" and m.acl and m.acl.permits(who)]
    # 3. defence in depth: re-check the live manifest, because the index copy of an ACL can be stale
    return kept, len(found) - len(kept)

from evidencedesk.ingest import naive
from evidencedesk.ingest.lab import HARBOR

shared = naive.shared_index(kb)  # the anti-pattern: one index for every tenant
q = "Who pays for repairs after damage?"
post, total = naive.post_filter_search(kb, shared, HARBOR, q, 3)
print("post-filter:", post, "| 'matches' shown to Omar:", total)
print("pre-filter: ", [h.chunk_id for h in search(kb, HARBOR, q, 3).hits])
print("user dict wins:", naive.merged_filter_search(kb, shared, HARBOR, "return a bike", {"tenant": "kestrel"}))

post-filter: [] | 'matches' shown to Omar: 16
pre-filter:  ['hb-returns:c0']
user dict wins: ['kc-returns:c0', 'kc-returns-faq:c1', 'kc-fleet-policy:c0']

from pydantic import ValidationError

from evidencedesk.ingest.access import Filters

for bad in ({"tenant": "kestrel"}, {"allow": ["tenant:kestrel"]}):
    try:
        Filters(**bad)
    except ValidationError as error:
        print("refused:", error.errors()[0]["type"], error.errors()[0]["loc"])
asked = search(kb, HARBOR, "return a bike", 5, Filters(sources=("kc-returns",)))
print("Omar names a Kestrel source:", asked.hits, repr(asked.message))

refused: extra_forbidden ('tenant',)
refused: extra_forbidden ('allow',)
Omar names a Kestrel source: () 'No matching documents.'

from evidencedesk.ingest.lab import STAFF

Q = "goodwill credit limit"


def goodwill_for(person):
    return sorted({h.source_id for h in search(kb1, person, Q, 3).hits if h.source_id == "kc-goodwill"})


kb1, _ = fresh("v1")
cached = CachedSearch(kb1)
print("v1  staff:", goodwill_for(STAFF), "| lead:", goodwill_for(LEAD), "| cache:", cached.ask(STAFF, Q)[1], cached.ask(STAFF, Q)[1])
change = kb1.ingest(snapshot("v2"), prune=True)  # the goodwill note now allows group:team-leads only
print("v2  staff:", goodwill_for(STAFF), "| lead:", goodwill_for(LEAD), "| cache:", cached.ask(STAFF, Q)[1])
print("goodwill:", {o.source_id: o.outcome for o in change.outcomes}["kc-goodwill"], "| no re-chunking for it: version", kb1.manifests["kc-goodwill"].version)

v1  staff: ['kc-goodwill'] | lead: ['kc-goodwill'] | cache: miss hit
v2  staff: [] | lead: ['kc-goodwill'] | cache: miss
goodwill: acl_updated | no re-chunking for it: version 1

from evidencedesk.ingest.contract import Acl
from evidencedesk.ingest.verify import verify

before = kb1.late_denied
kb1.parts["kestrel"].acl["kc-payroll-note:c0"] = Acl("kestrel", frozenset({"tenant:kestrel"}), "public")  # a stale, too-loose index copy
hits = search(kb1, CUSTOMER, "payroll cut-off salary", 5).hits
print("payroll chunk returned:", any(h.source_id == "kc-payroll-note" for h in hits), "| late denials:", kb1.late_denied - before)
print("verify:", verify(kb1))
kb1.ingest(snapshot("v2"), prune=True)  # the sources did not change, yet index copies are derived again
print("after a re-run:", verify(kb1))

payroll chunk returned: False | late denials: 1
verify: [Finding(code='acl_drift', subject='kc-payroll-note:c0')]
after a re-run: []

only_kestrel = [e for e in snapshot("v1") if e["tenant"] == "kestrel"]
a, b = KnowledgeBase(), KnowledgeBase()
a.ingest(only_kestrel)
b.ingest(snapshot("v1"))  # the same plus Harbor's documents
q = "How long do I have to return a bike?"
print("scores identical with and without the other tenant:", search(a, CUSTOMER, q, 5) == search(b, CUSTOMER, q, 5))
c = KnowledgeBase()
c.ingest([e for e in only_kestrel if e["source_id"] != "kc-payroll-note"])
raw = lambda base: round(base.parts["kestrel"].hybrid.sparse.bm25.score(q, "kc-returns:c0"), 4)
print("raw BM25 of one chunk with / without a hidden HR note:", raw(a), raw(c))

scores identical with and without the other tenant: True
raw BM25 of one chunk with / without a hidden HR note: 1.8298 1.765

python -m pytest tests/test_ingest_pipeline.py tests/test_ingest_access.py tests/test_ingest_erasure.py tests/test_ingest_rewrite.py | tail -n 1 | sed 's/ in [0-9.]*s//'

44 passed

from evidencedesk.observe.events import EventLog
from evidencedesk.usage import ManualClock

log = EventLog(ManualClock(1000.0))
kb3, _ = fresh("v1", log=log)
search(kb3, CUSTOMER, "salary review and payroll", 3, request_id="req-1")
for e in log.records():
    if e["event"] in ("ingest.run_end", "search.query"):
        print(e["event"], e["user"] or "-", e["fields"])

ingest.run_end - {'embedded': 18, 'epoch': 1, 'findings': 0, 'n_added': 11, 'n_quarantined': 1, 'run_id': 'run-001'}
search.query u-d14b511a {'hits': 3, 'k': 3, 'late_denied': 0, 'query_chars': 25}

from evidencedesk.ingest.rewrite import DOMAIN, StripRewriter, TableRewriter, compare, compress_report, load_cases, outcome

cases = load_cases()
print(len(cases), "fixed questions | hit@1 of the original questions:", round(sum(outcome(kb1, c, c.question, 1) for c in cases) / len(cases), 3))
for r in (DOMAIN, StripRewriter(), TableRewriter("greedy", (("return", "return salary reviews payroll goodwill"),))):
    got = compare(kb1, cases, r, 1)
    print(f"{r.name:<7} {got.original} -> {got.rewritten} | better {list(got.delta.better)} worse {list(got.delta.worse)} p={got.delta.p} | leaks {got.leaks}")

15 fixed questions | hit@1 of the original questions: 0.733
domain  0.733 -> 0.867 | better ['q05', 'q07'] worse [] p=0.5 | leaks 0
strip   0.733 -> 0.667 | better [] worse ['q15'] p=1.0 | leaks 0
greedy  0.733 -> 0.733 | better [] worse [] p=1.0 | leaks 0

for per_chunk in (1, 2):
    got = compress_report(kb1, cases, per_chunk, 3)
    print(f"{per_chunk} sentence(s) per chunk: {got.tokens_before} -> {got.tokens_after} tokens ({1 - got.tokens_after / got.tokens_before:.0%} saved), evidence lost in {list(got.lost)}")

1 sentence(s) per chunk: 1107 -> 609 tokens (45% saved), evidence lost in ['q03', 'q12']
2 sentence(s) per chunk: 1107 -> 985 tokens (11% saved), evidence lost in ['q03']

entry = dict(next(e for e in snapshot("v1") if e["source_id"] == "kc-warranty"), tenant="kestrl")
print(KnowledgeBase().ingest([entry]).outcomes[0])

Outcome(source_id='kc-warranty', outcome='quarantined', version=0, chunks=0, stage='classify', reasons=('unknown_tenant', 'acl_too_broad'))

from evidencedesk.ingest.contract import Principal

entries = [dict(e) for e in snapshot("v2")]
next(e for e in entries if e["source_id"] == "kc-fleet-policy")["allow"] = ["group:workshop"]
step = kb1.ingest(entries, prune=True)
shop = Principal("priya", "kestrel", ("workshop",))
print({o.source_id: o.outcome for o in step.outcomes if o.outcome != "unchanged"}, "| embedded", step.embedded)
for person in (STAFF, shop):
    print(person.user_id, sorted({h.source_id for h in search(kb1, person, "borrow a demo bike", 3).hits if h.source_id == "kc-fleet-policy"}))

{'kc-fleet-policy': 'acl_updated', 'kc-promo': 'quarantined'} | embedded 0
aiko []
priya ['kc-fleet-policy']

import shutil
import tempfile
from pathlib import Path

from evidencedesk.ingest.contract import DATA

with tempfile.TemporaryDirectory() as folder:
    work = Path(folder)
    shutil.copytree(DATA / "files", work / "files")
    kb4 = KnowledgeBase(root=work)
    kb4.ingest(snapshot("v1"))
    warranty = work / "files/kc-warranty.csv"
    warranty.write_text(warranty.read_text(encoding="utf-8").replace("battery,24", "battery,36"), encoding="utf-8")
    step = kb4.ingest([e for e in snapshot("v1") if e["source_id"] != "kc-stores"], prune=True)  # kc-stores vanished
    print({o.source_id: o.outcome for o in step.outcomes if o.outcome != "unchanged"}, "| embedded", step.embedded, "| findings", step.findings)
    hit = search(kb4, CUSTOMER, "How many months is the battery warranty?", 1).hits[0]
    print(hit.chunk_id, "| row says 36:", "| battery | 36 |" in hit.snippet)
    stores = search(kb4, CUSTOMER, "Which store has a workshop in Dhaka?", 3).hits
    print([h.source_id for h in stores], "| Dhaka in text:", any("Dhaka" in h.snippet for h in stores), "| tombstone:", kb4.manifests["kc-stores"].status)

{'kc-warranty': 'updated', 'kc-promo': 'quarantined', 'kc-stores': 'deleted'} | embedded 1 | findings []
kc-warranty:c0 | row says 36: True
['kc-warranty', 'kc-warranty', 'kc-returns'] | Dhaka in text: False | tombstone: deleted