mixle.substrate package

Typed, provenanced, scoped storage for local knowledge and artifacts.

The substrate stores raw data, documents, model artifacts, harvested traces, ontology triples, simulation outputs, and context packets as SubstrateItem objects. Each item carries a kind, provenance, access scope, tags, links, and a retrievable text surface.

class Substrate(root=None)[source]

Bases: object

A local shard of the knowledge substrate: a filesystem-backed store with typed retrieval.

put / get / remove / all manage items; search retrieves the k most relevant items for a query, filtered by kind and scope, ranking text items semantically (a learned embedding over the current text corpus) and everything else lexically. save / load persist the shard as one items.jsonl under root.

Parameters:

root (str | None)

put(item)[source]

Add or replace an item; returns its id. Marks the semantic index dirty for text items.

Parameters:

item (SubstrateItem)

Return type:

str

add(kind, text='', **kw)[source]

Convenience: build a SubstrateItem and put() it.

Parameters:
Return type:

str

get(item_id)[source]
Parameters:

item_id (str)

Return type:

SubstrateItem | None

remove(item_id)[source]
Parameters:

item_id (str)

Return type:

bool

all(*, kind=None, scope=None)[source]
Parameters:

  • kind (str | None)

  • scope (str | None)

Return type:

list[SubstrateItem]

reindex()[source]

(Re)fit the embedding index over the current text-bearing items. Idempotent, lazy-called.

Return type:

None

search(query, k=5, *, kind=None, scope=None)[source]

The k most relevant items to query as (item, score), filtered by kind/scope.

Text-bearing items rank by cosine similarity in the learned embedding space; when there are too few items to learn one (or for a non-text query), ranking falls back to a lexical token overlap. Structured items with no text always rank lexically over their serialized payload + tags.

Parameters:
Return type:

list[tuple[SubstrateItem, float]]

save(root=None)[source]

Persist the shard to {root}/items.jsonl (one item per line).

Parameters:

root (str | None)

Return type:

str

load(root=None)[source]
Parameters:

root (str | None)

Return type:

None

class SubstrateItem(kind, text='', payload=<factory>, provenance=<factory>, scope='local', tags=<factory>, links=<factory>, id=<factory>, created_at=<factory>)[source]

Bases: object

One typed, provenanced, scoped item in the substrate.

Parameters:
kind: str
text: str = ''
payload: dict[str, Any]
provenance: dict[str, Any]
scope: str = 'local'
tags: list[str]
links: list[str]
id: str
created_at: float
to_json()[source]
Return type:

dict[str, Any]

classmethod from_json(d)[source]
Parameters:

d (dict[str, Any])

Return type:

SubstrateItem

ingest_documents(substrate, docs, *, source='documents', scope='local')[source]

Add text passages to the substrate as kind="text" items. Returns the new item ids.

Each doc is a string, or a {"text": ..., "tags": [...], "payload": {...}} dict for metadata.

Parameters:
Return type:

list[str]

ingest_artifacts(substrate, registry_root, *, scope='local')[source]

Index every deployed artifact under registry_root (dirs containing a manifest.json).

The item’s text surface is a human summary of the manifest (kind, io, meta); its payload REFERENCES the artifact directory ({"ref": path}) rather than copying it, and provenance carries the manifest’s lineage fields when present.

Parameters:

  • substrate (Substrate)

  • registry_root (str)

  • scope (str)

Return type:

list[str]

ingest_traces(substrate, jsonl_path, *, source=None, scope='local')[source]

Index a harvested .jsonl of {"input": ..., "answer"/"label"/"call": ...} rows as traces.

Parameters:

  • substrate (Substrate)

  • jsonl_path (str)

  • source (str | None)

  • scope (str)

Return type:

list[str]

ingest_file(substrate, path, *, kind=None, source=None, scope='local')[source]

Ingest a data file into the substrate. Format inferred from the extension unless kind forces it.

.txt/.md -> one text item per non-empty line; .jsonl -> one item per JSON line (a string / {"text": ...} becomes a text item, any other object a record item); .csv -> one record item per row keyed by the header. source defaults to the file path.

Parameters:

  • substrate (Substrate)

  • path (str)

  • kind (str | None)

  • source (str | None)

  • scope (str)

Return type:

list[str]

ingest_records(substrate, records, *, source='records', scope='local', text_fields=None)[source]

Add a sequence of records (dicts or tuples) to the substrate as kind="record" items.

Each record’s payload is stored structured; its retrievable text surface is the text_fields (for dict records) joined, else the whole serialized record – so records are searchable by content.

Parameters:
Return type:

list[str]

class ContextPacket(task, items=<factory>, scores=<factory>, budget=<factory>, used_chars=0, n_candidates=0, texts=<factory>, compressed=False)[source]

Bases: object

A budgeted, provenanced view of the substrate assembled for one target + task.

texts holds the text actually used per item – the full item surface, or (when the packet was compressed) an extractive summary that keeps only the query-relevant sentences. preservation receipts how much of each item’s query-relevant content survived, so compression is measured, not trusted.

Parameters:
task: str
items: list[SubstrateItem]
scores: list[float]
budget: ContextBudget
used_chars: int = 0
n_candidates: int = 0
texts: list[str]
compressed: bool = False
render(*, header=True)[source]

The assembled context string the target consumes (respecting the budget shape).

Parameters:

header (bool)

Return type:

str

preservation()[source]

Per item, the fraction of the task’s query terms retained in the used text (1.0 = all kept).

The receipt for compression: a value near 1.0 means the summary kept what the query cares about; a low value flags an item whose relevant content was squeezed out.

Return type:

list[float]

property compression_ratio: float

Used chars / full chars over the selected items (1.0 = uncompressed).

provenance()[source]

Where every included piece came from – ids, kinds, sources, relevance scores.

Return type:

list[dict[str, Any]]

as_dict()[source]
Return type:

dict[str, Any]

class ContextBudget(max_chars=2000, max_items=20, shape='passages')[source]

Bases: object

What a target can take – the DeviceSpec of context. shape hints the rendering style.

Parameters:

  • max_chars (int)

  • max_items (int)

  • shape (str)

max_chars: int = 2000
max_items: int = 20
shape: str = 'passages'
assemble_context(substrate, task, *, budget=None, kind=None, scope=None, compress=False, telemetry=None)[source]

Assemble the best-affordable ContextPacket for task from substrate.

Retrieves relevant items (Substrate.search()), then packs them in descending relevance until the character budget or item cap is reached – always keeping at least the single most relevant item so a tiny budget still yields something. With compress=True, an item too large to fit whole is extractively summarized to its query-relevant sentences instead of dropped; packet.preservation() reports what was kept. Emits a context event when telemetry is supplied.

Parameters:

  • substrate (Substrate)

  • task (str)

  • budget (ContextBudget | None)

  • kind (str | None)

  • scope (str | None)

  • compress (bool)

  • telemetry (Any)

Return type:

ContextPacket

compress_text(text, task, max_chars)[source]

Extractive, torch-free summary of text keeping the sentences most relevant to task, within max_chars (the standalone compressor used by assemble_context() with compress=True).

Parameters:
Return type:

str

retrieve(substrate, query, *, k=8, kinds=None, weights=None, diversify=True, scope=None, telemetry=None)[source]

Plan a cross-kind retrieval for query (see module docstring).

Parameters:

  • k (int) – total items to return.

  • kinds (list[str] | None) – restrict to these substrate kinds (default: every kind present).

  • weights (dict[str, float] | None) – per-kind score multipliers (e.g. {"artifact": 1.2} to favor deployable models).

  • diversify (bool) – when True (default), interleave the top hits of each kind so the result spans modalities; when False, take a flat merged top-k (whichever kind scores highest wins).

  • scope (str | None) – restrict to a team/access scope.

  • substrate (Substrate)

  • query (str)

  • telemetry (Any)

Return type:

Retrieval

class Retrieval(query, items=<factory>, scores=<factory>)[source]

Bases: object

A planned, cross-kind retrieval result: items in merged relevance order, grouped by kind.

Parameters:
query: str
items: list[SubstrateItem]
scores: list[float]
by_kind()[source]
Return type:

dict[str, list[SubstrateItem]]

kinds()[source]
Return type:

list[str]

top(n)[source]
Parameters:

n (int)

Return type:

list[SubstrateItem]

provenance()[source]
Return type:

list[dict[str, Any]]

to_context(task=None, **assemble_kw)[source]

Assemble a ContextPacket from this retrieval (over an in-memory shard of its items).

Parameters:

  • task (str | None)

  • assemble_kw (Any)

Return type:

Any

multihop(substrate, query, *, max_hops=2, seeds=3, branch=2, max_items=12, min_score=0.0, scope=None, telemetry=None)[source]

Chain typed hops from query across the substrate, recording the evidence path (see docstring).

Parameters:

  • max_hops (int) – how many hops out from the seeds to expand.

  • seeds (int) – how many top matches to start from (depth 0).

  • branch (int) – how many neighbors to expand per frontier item per hop.

  • max_items (int) – overall cap on the chain size.

  • min_score (float) – relevance floor for SEED and CONTENT hops – a match must score strictly above it to enter the chain (LINK hops are explicit edges and always followed). Keeps a fuzzy retriever from chaining on near-zero-similarity noise; raise it for a dense embedder.

  • scope (str | None) – restrict to a team/access scope.

  • substrate (Substrate)

  • query (str)

  • telemetry (Any)

Return type:

HopChain

class HopChain(query, steps=<factory>)[source]

Bases: object

A multi-hop retrieval result: the items found and the evidence PATH to each.

Parameters:

  • query (str)

  • steps (list[HopStep])

query: str
steps: list[HopStep]
property items: list[SubstrateItem]
by_depth()[source]
Return type:

dict[int, list[SubstrateItem]]

max_depth()[source]
Return type:

int

path_to(item_id)[source]

The evidence chain from a seed to item_id – the trace the reasoner cites.

Parameters:

item_id (str)

Return type:

list[SubstrateItem]

provenance()[source]
Return type:

list[dict[str, Any]]

to_context(task=None, **assemble_kw)[source]
Parameters:

  • task (str | None)

  • assemble_kw (Any)

Return type:

Any

class HopStep(item, depth, via, parent_id, score=0.0)[source]

Bases: object

One item in the chain plus how it was reached: the provenance of a retrieval decision.

Parameters:

  • item (SubstrateItem)

  • depth (int)

  • via (str)

  • parent_id (str | None)

  • score (float)

item: SubstrateItem
depth: int
via: str
parent_id: str | None
score: float = 0.0
answer_from_substrate(substrate, question, answerer, *, budget=None, hops=1, min_evidence=1, min_confidence=0.1, compress=True, scope=None, telemetry=None)[source]

Answer question from substrate via answerer, or abstain when evidence is too thin.

Parameters:

  • answerer (Callable[[str, str], str]) – (question, context_text) -> answer_str – any model/rule; called ONLY when there is enough evidence above the confidence floor (so a weak retrieval never fabricates).

  • budget (ContextBudget | None) – the context budget handed to the answerer (default 2000 chars).

  • hops (int) – 1 = single-shot retrieve(); >1 = multihop() chaining that many hops.

  • min_evidence (int) – minimum retrieved items required to attempt an answer.

  • min_confidence (float) – retrieval-strength floor below which it abstains rather than guess.

  • compress (bool) – compress the context to fit more sources under budget.

  • scope (str | None) – restrict to a team/access scope.

  • substrate (Substrate)

  • question (str)

  • telemetry (Any)

Return type:

Answer

class Answer(question, answer, abstained, confidence, context, note='', evidence=<factory>)[source]

Bases: object

A cited answer or abstention with the evidence it rests on and a confidence.

Parameters:

  • question (str)

  • answer (str | None)

  • abstained (bool)

  • confidence (float)

  • context (ContextPacket)

  • note (str)

  • evidence (list[SubstrateItem])

question: str
answer: str | None
abstained: bool
confidence: float
context: ContextPacket
note: str = ''
evidence: list[SubstrateItem]
citations()[source]

Where the answer’s evidence came from – the provenance the answer must be checkable against.

Return type:

list[dict[str, Any]]

as_dict()[source]
Return type:

dict[str, Any]

investigate(question, actions, answerer, *, budget_cost=None, min_evidence=1, min_confidence=0.15, target_confidence=None, max_actions=None, scorer=None, telemetry=None)[source]

Answer question by firing evidence-acquiring actions under a cost budget, or abstain.

Actions are ordered by scorer (default score_action(), EIG-per-cost) and fired highest-first. The loop stops early once it holds at least min_evidence fragments and confidence clears target_confidence (default: min_confidence). It also stops at the cost budget or max_actions. scorer can be replaced with a learned acquisition policy such as mixle.inference.learn_action_policy(). The answerer is called only when the evidence clears the bar. The returned Investigation carries the ordered action trace as provenance, and each fired action can emit telemetry for later policy learning.

Parameters:
Return type:

Investigation

class Investigation(question, answer, abstained, confidence, steps=<factory>, note='', factuality=None, proposal=None)[source]

Bases: object

A cited answer (or abstention) plus the sequence of actions that acquired its evidence.

Parameters:
question: str
answer: str | None
abstained: bool
confidence: float
steps: list[Step]
note: str = ''
factuality: Any = None
proposal: Any = None
property evidence: list[str]
property spent: float
trace()[source]

The actions taken, in order – the provenance the answer must be checkable against.

Return type:

list[dict[str, Any]]

as_dict()[source]
Return type:

dict[str, Any]

class Action(name, kind, run, cost=1.0, description='', base_score=0.0)[source]

Bases: object

One evidence-acquiring move: run it on a question, get back evidence fragments, at a cost.

Parameters:
name: str
kind: str
run: Callable[[str], list[str]]
cost: float = 1.0
description: str = ''
base_score: float = 0.0
class Step(action, kind, fragments, cost, score, relevance=0.0)[source]

Bases: object

A fired action and what it yielded – the audit trail behind an investigated answer.

Parameters:
action: str
kind: str
fragments: list[str]
cost: float
score: float
relevance: float = 0.0
score_action(action, question)[source]

EIG-per-cost proxy: lexical relevance of the action to the question, divided by its cost.

This heuristic can be replaced with a learned or calibrated expected-information-gain estimate. Retrieval-style actions carry a base_score floor because retrieval is always at least weakly informative.

Parameters:

  • action (Action)

  • question (str)

Return type:

float

relevance_of(action, question)[source]

How on-topic an action is for a question (lexical overlap + its base floor), ignoring cost.

Parameters:

  • action (Action)

  • question (str)

Return type:

float

action_features(action, question)[source]

The features a learned acquisition policy keys on: the action’s kind, cost, and query overlap.

Parameters:

  • action (Action)

  • question (str)

Return type:

dict[str, Any]

retrieve_action(substrate, *, name='retrieve', k=6, scope=None, cost=1.0, min_score=0.0)[source]

A RETRIEVE action over a Substrate (the always-available floor action).

min_score filters out weak matches: a tiny embedder returns SOMETHING for every query, so a positive floor keeps genuinely-irrelevant items from becoming false evidence. It defaults to 0.0 (keep everything) but a small positive value makes retrieval honest on a noisy index.

Parameters:
Return type:

Action

compute_action(skill, *, name=None, cost=1.0, description=None)[source]

A COMPUTE action that runs a Skill and reports its result.

Parameters:
Return type:

Action

simulate_action(simulator, field_index, scenario, *, name=None, cost=2.0, description='')[source]

A SIMULATE action that runs a what-if scenario and reports the simulated outcome mean.

Parameters:
Return type:

Action

create_action(build, *, name='create', cost=4.0, description='', report=None)[source]

A CREATE action that builds a model / dataset on demand and reports what it made.

build(question) -> artifact fits/synthesizes something (e.g. via mixle.inference.create() or mixle.inference.synthesize()); report renders the artifact to an evidence fragment (default: a certificate/guarantee summary when present, else repr). Creation is the most expensive action, so it defaults to cost=4 – the reasoner reaches for it only when cheaper retrieve/compute/simulate actions cannot answer.

Parameters:
Return type:

Action

delegate_action(delegate, *, name='delegate', cost=8.0, description='', priced=True)[source]

A DELEGATE action that hands the question to an external worker (pool job / remote tool / agent).

delegate(question) -> answer is any priced external capability. This is the reasoner’s most expensive move (default cost=8) and the escalation of last resort: it fires only when nothing local clears the bar, honoring the 99%-local topology. priced=True records that the call incurs real spend (the pool/interop layers own the actual budget-reject + confirm rails).

Parameters:
Return type:

Action

class Reasoner(answerer, *, substrate=None, skills=None, actions=None, budget_cost=None, min_confidence=0.15, retrieve_min_score=0.0, scorer=None, telemetry=None)[source]

Bases: object

A configured reasoner: a knowledge store + skills + actions, asked questions through one method.

Parameters:

  • answerer (Callable[[str, str], str])

  • substrate (Substrate | None)

  • skills (Any)

  • actions (list[Action] | None)

  • budget_cost (float | None)

  • min_confidence (float)

  • retrieve_min_score (float)

  • scorer (Callable[[Action, str], float] | None)

  • telemetry (Any)

property actions: list[Action]

The current action space (retrieve + one compute per skill + attached simulators/creators).

add_action(action)[source]

Attach an extra action (a simulator, creator, or delegate); returns self for chaining.

Parameters:

action (Action)

Return type:

Reasoner

use_policy(scorer)[source]

Route by a learned acquisition policy instead of the lexical prior. Chainable.

Parameters:

scorer (Callable[[Action, str], float])

Return type:

Reasoner

ask(question, *, verify=False, **overrides)[source]

Answer question over the configured action space, or abstain. overrides pass through to investigate() (e.g. budget_cost, min_confidence, target_confidence, max_actions).

With verify=True and a substrate configured, the (non-abstained) answer is run back through check_factuality() and the FactualityReceipt is attached to Investigation.factuality – the reasoner grounds its own answer’s claims and reports which it can cite. It does not suppress the answer; the receipt is there for the caller to gate on.

Parameters:
Return type:

Investigation

check_factuality(substrate, answer, *, extract=None, corroborates=None, min_score=0.2, k=4, scope=None)[source]

Ground each claim of answer against substrate, returning a FactualityReceipt.

Parameters:

  • extract (Callable[[str], list[str]] | None) – answer -> [claim, ...] (default mixle.reason.llm.sentence_claims()).

  • corroborates (Callable[[str, str], bool] | None) – (evidence_text, claim) -> bool deciding if retrieved evidence supports a claim (default content-overlap; pass an NLI/entailment check for stronger grounding).

  • min_score (float) – retrieval-score floor; evidence below it doesn’t count (guards tiny-embedder noise).

  • k (int) – evidence items retrieved per claim.

  • scope (str | None) – restrict retrieval to a team/access scope.

  • substrate (Substrate)

  • answer (str)

Return type:

FactualityReceipt

class FactualityReceipt(answer, verdicts=<factory>)[source]

Bases: object

A per-claim grounding of an answer against the substrate – the receipt behind ‘is this true?’.

Parameters:

  • answer (str)

  • verdicts (list[ClaimVerdict])

answer: str
verdicts: list[ClaimVerdict]
property grounded_fraction: float
unsupported()[source]

The claims the substrate could not corroborate – exactly what to flag or retract.

Return type:

list[ClaimVerdict]

is_grounded(threshold=1.0)[source]

True iff the grounded fraction meets threshold (default 1.0: every claim must be cited).

Parameters:

threshold (float)

Return type:

bool

as_dict()[source]
Return type:

dict[str, Any]

class ClaimVerdict(claim, supported, score, citations=<factory>)[source]

Bases: object

One claim from an answer, marked supported or not, with the evidence that (dis)confirms it.

Parameters:
claim: str
supported: bool
score: float
citations: list[dict[str, Any]]
class Space(substrate, team, *, shared=(PUBLIC,))[source]

Bases: object

A team’s scoped view over a shared substrate: its own items plus what has been shared to it.

Parameters:

  • substrate (Substrate)

  • team (str)

  • shared (tuple[str, ...])

property scopes: set[str]
all(*, kind=None)[source]

Every visible item (optionally of one kind) – never another team’s private knowledge.

Parameters:

kind (str | None)

Return type:

list[SubstrateItem]

add(*, scope=None, **kw)[source]

Add an item to this team’s own scope by default (pass scope=PUBLIC to share immediately).

Parameters:
Return type:

str

retrieve(query, *, k=8, **kw)[source]

Retrieve over exactly the team’s visible set (own scope ∪ shared), with cross-kind diversity.

Parameters:
Return type:

Any

publish(ids, *, to=PUBLIC, by=None)[source]

Share this team’s items into a common scope (audited). Only own-scope items are publishable.

Parameters:
Return type:

list[str]

publish(substrate, ids, *, to=PUBLIC, by=None, from_scope=None)[source]

Share items into a common scope.

Re-scopes each item in ids to to and records published_by / published_from in its provenance. Returns the ids actually published (missing ids are skipped). from_scope, if given, guards that only items currently in that scope are published (an ACL check the caller can enforce).

Parameters:

  • substrate (Substrate)

  • ids (list[str])

  • to (str)

  • by (str | None)

  • from_scope (str | None)

Return type:

list[str]

visible_scopes(team, *, shared=(PUBLIC,))[source]

The scopes a team may read: its own id plus the shared scopes (never another team’s private one).

Parameters:
Return type:

set[str]

merge_versions(substrate, keep_id, other_id, *, by=None, prefer='latest')[source]

Reconcile two versions of the same knowledge into one, keeping full lineage (no silent loss, P2).

Merges other_id into keep_id: unions tags and links, keeps the text/payload of whichever has the higher version (prefer="latest") or of keep (prefer="keep"), bumps the surviving item’s version, records BOTH parents in the history, and removes the merged-away item. Returns the surviving id, or None if either is missing. Two teams that independently edited a shared item can be reconciled without either edit vanishing unrecorded.

Parameters:

  • substrate (Substrate)

  • keep_id (str)

  • other_id (str)

  • by (str | None)

  • prefer (str)

Return type:

str | None

history(substrate, item_id)[source]

The full publish history of an item: every version with who shared it, from where, to where.

Parameters:

  • substrate (Substrate)

  • item_id (str)

Return type:

list[dict[str, Any]]

version_of(item)[source]

The share version of an item (0 if never published) – a monotonic counter bumped by each publish.

Parameters:

item (Any)

Return type:

int

verify_lineage(substrate, item_id, *, max_depth=20)[source]

Walk item_id’s ancestry via links, reporting dangling edges and intact depth (cycle-safe).

An item is intact iff every lineage link, transitively, resolves to an item that exists. Cycles are handled (each id is visited once). max_depth bounds pathological chains. A missing root item yields intact=False with itself recorded as dangling.

Parameters:

  • substrate (Substrate)

  • item_id (str)

  • max_depth (int)

Return type:

LineageReport

audit_substrate(substrate, *, scope=None)[source]

A knowledge-integrity sweep: how many items have intact lineage, and every broken one named.

Returns {n_items, n_intact, n_broken, broken: [{item_id, dangling}, ...]} – the store’s trust surface at a glance, so a broken provenance edge surfaces as a finding rather than a silent lie.

Parameters:

  • substrate (Substrate)

  • scope (str | None)

Return type:

dict[str, Any]

class LineageReport(item_id, intact, n_links, dangling=<factory>, depth=0, visited=0)[source]

Bases: object

Whether an item’s provenance chain resolves end to end – and where it breaks if it doesn’t.

Parameters:
item_id: str
intact: bool
n_links: int
dangling: list[str]
depth: int = 0
visited: int = 0
as_dict()[source]
Return type:

dict[str, Any]

class Governance(approvers=<factory>)[source]

Bases: object

Who may approve promotions into which scope – the org-governance ACL.

Parameters:

approvers (dict[str, set[str]])

approvers: dict[str, set[str]]
may_approve(who, scope)[source]
Parameters:
Return type:

bool

grant(who, scope)[source]

Add who as an approver for scope (chainable).

Parameters:
Return type:

Governance

propose(substrate, ids, *, to, by=None)[source]

Mark items as pending promotion to scope to (they are NOT yet visible there). Returns the ids.

Parameters:
Return type:

list[str]

approve(substrate, item_id, *, by, governance, to=None)[source]

Promote a pending item into its proposed scope – IFF by may approve for that scope (the gate).

On success the item is published into the target scope (via P1 publish(), so it inherits the versioned/audited share) and its proposal is marked approved with the approver id. Returns False (no change) if the item has no pending proposal or by lacks approval rights.

Parameters:

  • substrate (Substrate)

  • item_id (str)

  • by (str)

  • governance (Governance)

  • to (str | None)

Return type:

bool

reject(substrate, item_id, *, by, reason='')[source]

Refuse a pending promotion – the item stays in its origin scope; the refusal is recorded.

Parameters:

  • substrate (Substrate)

  • item_id (str)

  • by (str)

  • reason (str)

Return type:

bool

pending(substrate, *, to=None)[source]

Items awaiting approval (optionally only those proposed to scope to).

Parameters:

  • substrate (Substrate)

  • to (str | None)

Return type:

list[SubstrateItem]

detect_secrets(text)[source]

Scan text for well-known secret shapes; return a SecretScan naming each finding.

Parameters:

text (str)

Return type:

SecretScan

redact_secrets(text, *, mask='[REDACTED:{rule}]', keep_prefix=0)[source]

Return text with every detected secret replaced by a rule-labelled mask (destructive to secrets).

keep_prefix leaves that many leading characters of the secret visible (0 = fully masked) so a reader can still tell which credential it was without recovering it.

Parameters:
Return type:

str

safe_text(text)[source]

Redact-before-store guard: mask any secrets so they are never indexed or served.

Parameters:

text (str)

Return type:

str

scan_item(item)[source]

Scan a substrate item’s text surface for secrets.

Parameters:

item (Any)

Return type:

SecretScan

scan_substrate(substrate, *, scope=None)[source]

Sweep a substrate for leaked secrets: which items are dirty and which rules tripped.

Returns {n_items, n_dirty, dirty: [{item_id, rules}]} – a leak surface at a glance, so a pasted key or a token in a stored trace surfaces as a finding rather than sitting indexed and searchable.

Parameters:

  • substrate (Any)

  • scope (str | None)

Return type:

dict[str, Any]

class SecretScan(findings=<factory>)[source]

Bases: object

The result of scanning a text: whether anything leaked and every finding.

Parameters:

findings (list[SecretFinding])

findings: list[SecretFinding]
property clean: bool
rules()[source]
Return type:

list[str]

as_dict()[source]
Return type:

dict[str, Any]

class SecretFinding(rule, start, end, preview)[source]

Bases: object

One detected secret: which rule matched, where, and a safe preview (the value stays masked).

Parameters:
rule: str
start: int
end: int
preview: str
class ExternalModel(generate, *, calibration_prompts=None, equivalent=None, max_entropy=None, alpha=0.1, samples=8)[source]

Bases: object

An external generate callable wrapped so each answer carries semantic-entropy UQ.

Parameters:

  • generate (Callable[[Any], Any]) – prompt -> answer (an external agent / LLM / remote tool). Called multiple times per query to measure how much its meaning varies (the uncertainty signal).

  • calibration_prompts (Any) – optional example prompts; the (1-alpha) quantile of their semantic entropy becomes the “too uncertain” cutoff. Without them, max_entropy must be given (or every answer is treated as confident).

  • equivalent (Callable[[Any, Any], bool] | None) – (a, b) -> bool meaning-equivalence for clustering samples (default: exact match).

  • max_entropy (float | None) – an explicit uncertainty cutoff, overriding the calibrated one.

  • samples (int) – how many resamples to draw when measuring entropy.

  • alpha (float)

property max_entropy: float
answer(prompt)[source]

Call the external model and attach its semantic-entropy UQ (confident iff below the cutoff).

Parameters:

prompt (Any)

Return type:

ExternalAnswer

confident(prompt)[source]
Parameters:

prompt (Any)

Return type:

bool

class ExternalAnswer(prompt, answer, entropy, confident)[source]

Bases: object

An external model’s answer plus its self-measured uncertainty (semantic entropy).

Parameters:
prompt: Any
answer: Any
entropy: float
confident: bool
external_action(model, *, name='external', cost=8.0, description='', trust_uncertain=False)[source]

A reasoner DELEGATE action backed by a UQ-wrapped external model (see module docstring).

By default (trust_uncertain=False) the action contributes evidence ONLY when the external model is confident about the query; an uncertain external answer yields no fragment, so the reasoner treats it as no answer rather than a guess. The fragment carries the model’s entropy so the trace records how sure the external source was. Cost defaults high – external calls are the escalation of last resort.

Parameters:

  • model (ExternalModel)

  • name (str)

  • cost (float)

  • description (str)

  • trust_uncertain (bool)

Return type:

Any

kg_action(triples, *, ontology=None, types=None, name='kg', cost=1.0, description='', k=8)[source]

A reasoner RETRIEVE action over a knowledge graph (typed facts, not passages).

Contributes one fragment per fact (head relation tail); nothing links -> no evidence, so the reasoner falls through honestly instead of forcing a match. Relevance comes from the action’s description plus the KG’s own entity inventory (queries naming a known entity score).

Parameters:
Return type:

Any

link_entities(question, entities)[source]

The entity-linking leaf: which KG entities does the question mention?

Matches each entity’s normalized name as a token subsequence of the question, longest name first so multi-word entities win over their substrings. Returns the linked entities in match order.

Parameters:

  • question (str)

  • entities (Any)

Return type:

list[str]

retrieve_triples(triples, question, *, ontology=None, types=None, k=8)[source]

Typed KG retrieval: link the question’s entities, return the (schema-valid) facts about them.

Returns {entities, facts, rejected} – facts are the triples touching a linked entity (head or tail), at most k; when an ontology (+ entity types) is supplied, schema-violating triples are excluded and reported under rejected with named reasons, so a dirty store cannot inject a type-invalid fact as evidence.

Parameters:
Return type:

dict[str, Any]

check_freshness(substrate, item_id, *, max_age_s=None, now=None)[source]

Audit one item for the three staleness signals (see module docstring). Missing item -> stale.

Parameters:

  • substrate (Substrate)

  • item_id (str)

  • max_age_s (float | None)

  • now (float | None)

Return type:

Freshness

freshness_report(substrate, *, max_age_s=None, scope=None, now=None)[source]

Sweep a store for stale knowledge: {n_items, n_fresh, n_stale, stale: [...]} – the monitor feed.

Parameters:

  • substrate (Substrate)

  • max_age_s (float | None)

  • scope (str | None)

  • now (float | None)

Return type:

dict[str, Any]

content_hash(path)[source]

The sha256 (first 32 hex) of a file’s bytes, or None if unreadable – record this at ingest.

Parameters:

path (str)

Return type:

str | None

class Freshness(item_id, fresh, signals=<factory>, age_s=0.0)[source]

Bases: object

One item’s freshness verdict: fresh iff no staleness signal fired; every signal named.

Parameters:
item_id: str
fresh: bool
signals: list[str]
age_s: float = 0.0
as_dict()[source]
Return type:

dict[str, Any]

class Harness(reasoner, *, name, description='', validate=None, allowed_kinds=None, escalate=None, min_confidence=0.15, on_result=None)[source]

Bases: object

Schema + whitelist + guardrails + escalation around a reasoner (see module docstring).

Parameters:

  • reasoner (Reasoner) – the configured Reasoner (answerer + substrate + skills + actions).

  • description (str) – identity, used by the registry.

  • validate (Callable[[str], str | None] | None) – (request) -> None | str – return an error string to REFUSE the request before any model runs (the input schema, as a callable so any validator plugs in).

  • allowed_kinds (tuple[str, ...] | None) – action kinds the reasoner may fire (whitelist; None = all).

  • escalate (Callable[[str, Any], str] | None) – (request, result) -> str – called on abstention; its return is the escalated answer handed back (e.g. a ticket id). None = abstentions surface as ‘escalated’ with no handler note.

  • min_confidence (float) – the answer bar (passed through to ask).

  • on_result (Callable[[HarnessResult], None] | None) – optional UI hook, called with every HarnessResult (fire-and-forget).

  • name (str)

  • description

handle(request)[source]

Run one request through every gate: schema -> guardrails -> reasoner -> escalation.

Parameters:

request (str)

Return type:

HarnessResult

class HarnessResult(status, answer=None, reason='', investigation=None, redactions=0)[source]

Bases: object

One request’s outcome: which gate decided (refused/answered/escalated), and the evidence.

Parameters:

  • status (str)

  • answer (str | None)

  • reason (str)

  • investigation (Any)

  • redactions (int)

status: str
answer: str | None = None
reason: str = ''
investigation: Any = None
redactions: int = 0
as_dict()[source]
Return type:

dict[str, Any]

support_triage_harness(substrate, answerer, *, escalate=None, max_chars=2000)[source]

Support triage: retrieve-only over the team’s knowledge, refuse empty/oversized requests, escalate anything the knowledge base cannot support – the canonical ‘never guess at a customer’.

Parameters:
Return type:

Harness

monitoring_harness(reasoner, *, escalate=None)[source]

Monitoring/alerting: compute + simulate allowed (run checks, what-ifs), no delegation out.

Parameters:
Return type:

Harness

register_harness(substrate, harness, *, scope='local')[source]

Index a harness on the substrate as a scoped artifact – discoverable and shareable (P-scoped).

Parameters:

  • substrate (Substrate)

  • harness (Harness)

  • scope (str)

Return type:

str

find_harnesses(substrate, query='', *, scope=None)[source]

Discover registered harnesses (optionally by query / scope). Returns their manifests.

Parameters:

  • substrate (Substrate)

  • query (str)

  • scope (str | None)

Return type:

list[dict[str, Any]]

Submodules