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)

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)

Add or replace an item; returns its id and schedules semantic-index rebuilds for text items.

Parameters:

item (SubstrateItem)

Return type:

str

add(kind, text='', **kw)

Convenience: build a SubstrateItem and put() it.

Parameters:

• kind (str)

• text (str)

• kw (Any)

Return type:

str

get(item_id)

Return an item by id, or None when it is absent.

Parameters:

item_id (str)

Return type:

SubstrateItem | None

remove(item_id)

Remove an item by id and return whether anything was deleted.

Parameters:

item_id (str)

Return type:

bool

all(*, kind=None, scope=None)

Return stored items, optionally filtered by kind and scope.

Parameters:

• kind (str | None)

• scope (str | None)

Return type:

list[SubstrateItem]

reindex()

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

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:

• query (str)

• k (int)

• kind (str | None)

• scope (str | None)

Return type:

list[tuple[SubstrateItem, float]]

save(root=None)

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

Parameters:

root (str | None)

Return type:

str

load(root=None)

Load items from {root}/items.jsonl into this shard.

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

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)

• tags (list[str])

• links (list[str])

• id (str)

• created_at (float)

to_json()

Return this item as a JSON-serializable dictionary.

Return type:

dict[str, Any]

classmethod from_json(d)

Build a substrate item from its serialized dictionary form.

Parameters:

d (dict[str, Any])

Return type:

SubstrateItem

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

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:

• substrate (Substrate)

• docs (Sequence[str | dict[str, Any]])

• source (str)

• scope (str)

Return type:

list[str]

ingest_artifacts(substrate, registry_root, *, scope='local')

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

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

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)

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:

• substrate (Substrate)

• records (Sequence[Any])

• source (str)

• scope (str)

• text_fields (Sequence[str] | None)

Return type:

list[str]

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

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)

• n_candidates (int)

• texts (list[str])

• compressed (bool)

render(*, header=True)

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

Parameters:

header (bool)

Return type:

str

preservation()

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

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

Return type:

list[dict[str, Any]]

as_dict()

Return a JSON-serializable context-packet summary.

Return type:

dict[str, Any]

to_knowledge_dict(*, id, project_id, target_kind, target_id=None, expected_output_schema=None, factuality=None)

Return a plain dict shaped like mixle_knowledge.contracts.ContextPacket.

The exported fields cover id, project_id, task, target_kind, target_id, token and byte budgets, evidence item identifiers, constraints, citations, expected_output_schema, and payload. Constructing a validated pydantic object is the receiving package’s responsibility; core mixle intentionally keeps this as a dependency-free dictionary so platform contract packages can depend on core rather than the reverse.

When factuality is a FactualityReceipt, it is included in payload["factuality"] so receivers can inspect grounding metadata before trusting the packet.

Parameters:

• id (str)

• project_id (str)

• target_kind (str)

• target_id (str | None)

• expected_output_schema (dict[str, Any] | None)

• factuality (Any)

Return type:

dict[str, Any]

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

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)

class ReceiverProfile(name, max_chars=2000, max_items=20, shape='passages', compress=False)

Bases: object

A named receiver’s capacity – what assemble_for_receivers() budgets and shapes for it.

A frontier LM and a local student are not the same target: the LM affords a large, prose-shaped context; the student needs a small, feature-shaped one. ReceiverProfile names that difference so it is set once per receiver, not re-derived ad hoc at every call site.

Parameters:

• name (str)

• max_chars (int)

• max_items (int)

• shape (str)

• compress (bool)

to_budget()

Convert this receiver profile to a context budget.

Return type:

ContextBudget

assemble_context(substrate, task, *, budget=None, kind=None, scope=None, compress=False, telemetry=None)

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 small 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

assemble_for_receivers(substrate, task, receivers, *, kind=None, scope=None, telemetry=None)

Assemble ONE task-conditioned ContextPacket per named receiver – the concrete receiver-conditioned compression path.

Two receivers reading the same substrate for the same task get genuinely different renderings: budget, shape, and, via compress, which sentences survive. The result is not the same blob truncated to fit each consumer.

packets = assemble_for_receivers(substrate, task, [

ReceiverProfile(“frontier_llm”, max_chars=2000, shape=”passages”), ReceiverProfile(“local_student”, max_chars=200, shape=”features”, compress=True),

]) packets[“frontier_llm”].render(), packets[“local_student”].render()

Parameters:

• substrate (Substrate)

• task (str)

• receivers (Sequence[ReceiverProfile])

• kind (str | None)

• scope (str | None)

• telemetry (Any)

Return type:

dict[str, ContextPacket]

compress_text(text, task, max_chars)

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:

• text (str)

• task (str)

• max_chars (int)

Return type:

str

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

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

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

Group retrieved items by substrate kind.

Return type:

dict[str, list[SubstrateItem]]

kinds()

Return the sorted substrate kinds present in the result.

Return type:

list[str]

top(n)

Return the top n retrieved items.

Parameters:

n (int)

Return type:

list[SubstrateItem]

provenance()

Return compact provenance records for retrieved items.

Return type:

list[dict[str, Any]]

to_context(task=None, **assemble_kw)

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

Parameters:

• task (str | None)

• assemble_kw (Any)

Return type:

Any

eig_retrieve(substrate, belief, evidence_fn, *, k=8, kind=None, scope=None)

Greedily pick up to k substrate items by expected posterior-entropy reduction against belief.

evidence_fn(item) turns a candidate item into whatever belief.update(...) expects (e.g. a per-hypothesis log-likelihood vector for a CategoricalBelief). Each round, every remaining candidate is scored by current_belief.entropy() - updated_belief.entropy(); the best-scoring item is taken, the running belief moves to its post-update state, and scoring repeats against the shrunk pool – so an item whose evidence is redundant with an already-picked item scores near zero on its next look, the direct fix for similarity retrieval pulling in near-duplicates. Items whose evidence_fn raises (no usable evidence) are skipped, not fatal. Returned as a Retrieval (query is a fixed marker, not a text query) so it composes with the same to_context/by_kind surface as cosine retrieval.

Parameters:

• substrate (Substrate)

• belief (BeliefState)

• evidence_fn (Callable[[SubstrateItem], Any])

• k (int)

• kind (str | None)

• scope (str | None)

Return type:

Retrieval

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

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

Bases: object

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

Parameters:

• query (str)

• steps (list[HopStep])

property items: list[SubstrateItem]

Retrieved items in hop-chain order.

by_depth()

Group retrieved items by hop depth.

Return type:

dict[int, list[SubstrateItem]]

max_depth()

Return the deepest hop reached by the chain.

Return type:

int

path_to(item_id)

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

Parameters:

item_id (str)

Return type:

list[SubstrateItem]

provenance()

Return compact provenance records for every hop step.

Return type:

list[dict[str, Any]]

to_context(task=None, **assemble_kw)

Assemble the hop-chain items into a context packet.

Parameters:

• task (str | None)

• assemble_kw (Any)

Return type:

Any

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

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)

answer_from_substrate(substrate, question, answerer, *, budget=None, hops=1, min_evidence=1, min_confidence=0.1, compress=True, scope=None, telemetry=None)

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

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

citations()

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

Return type:

list[dict[str, Any]]

as_dict()

Return a JSON-serializable answer with citations and confidence.

Return type:

dict[str, Any]

measure_flywheel(sub, questions, answer_fn, assimilate_batch, *, k=5, min_credence=None)

Measure answer_fn against questions before and after assimilate_batch(sub) adds a batch of calibrated beliefs (returning the ids it touched), with a THIRD measurement that excludes exactly those ids from retrieval – the attribution control. answer_fn and every other piece of the system stay fixed throughout: only the store’s content (and what it makes retrievable) changes.

Parameters:

• sub (Substrate)

• questions (Sequence[QAItem])

• answer_fn (Callable[[str, list[str]], str])

• assimilate_batch (Callable[[Substrate], list[str]])

• k (int)

• min_credence (float | None)

Return type:

FlywheelReport

class FlywheelReport(before, after, withheld, attribution_confirmed)

Bases: object

Before/after/withheld flywheel measurements with an attribution check.

Parameters:

• before (FlywheelMeasurement)

• after (FlywheelMeasurement)

• withheld (FlywheelMeasurement)

• attribution_confirmed (bool)

class FlywheelMeasurement(solve_rate, grounded_fraction)

Bases: object

Held-out answer quality and grounding rate for one flywheel measurement.

Parameters:

• solve_rate (float)

• grounded_fraction (float)

class QAItem(question, answer)

Bases: object

One held-out question: answer_fn is judged correct on it if it produces answer from the retrieved context alone.

Parameters:

• question (str)

• answer (str)

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

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:

• question (str)

• actions (list[Action])

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

• budget_cost (float | None)

• min_evidence (int)

• min_confidence (float)

• target_confidence (float | None)

• max_actions (int | None)

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

• telemetry (Any)

Return type:

Investigation

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

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)

• proposal (Any)

property evidence: list[str]

Flatten all evidence fragments collected by the investigation.

property spent: float

Total action cost spent by the investigation.

trace()

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

Return type:

list[dict[str, Any]]

as_dict()

Return a JSON-serializable investigation summary.

Return type:

dict[str, Any]

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

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)

• description (str)

• base_score (float)

class Step(action, kind, fragments, cost, score, relevance=0.0)

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)

score_action(action, question)

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)

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)

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)

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

min_score filters out weak matches: a small embedder returns a result 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 conservative on a noisy index.

Parameters:

• substrate (Any)

• name (str)

• k (int)

• scope (str | None)

• cost (float)

• min_score (float)

Return type:

Action

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

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

Parameters:

• skill (Any)

• name (str | None)

• cost (float)

• description (str | None)

Return type:

Action

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

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

Parameters:

• simulator (Any)

• field_index (int)

• scenario (str)

• name (str | None)

• cost (float)

• description (str)

Return type:

Action

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

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:

• build (Callable[[str], Any])

• name (str)

• cost (float)

• description (str)

• report (Callable[[Any], str] | None)

Return type:

Action

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

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:

• delegate (Callable[[str], Any])

• name (str)

• cost (float)

• description (str)

• priced (bool)

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)

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)

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

Parameters:

action (Action)

Return type:

Reasoner

use_policy(scorer)

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)

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:

• question (str)

• verify (bool)

• overrides (Any)

Return type:

Investigation

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

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 low-signal 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>)

Bases: object

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

Parameters:

• answer (str)

• verdicts (list[ClaimVerdict])

property grounded_fraction: float

Fraction of extracted claims supported by substrate evidence.

unsupported()

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

Return type:

list[ClaimVerdict]

is_grounded(threshold=1.0)

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

Parameters:

threshold (float)

Return type:

bool

as_dict()

Return a JSON-serializable factuality receipt.

Return type:

dict[str, Any]

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

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

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]

Scopes visible to this team space.

all(*, kind=None)

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)

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

Parameters:

• scope (str | None)

• kw (Any)

Return type:

str

retrieve(query, *, k=8, **kw)

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

Parameters:

• query (str)

• k (int)

• kw (Any)

Return type:

Any

publish(ids, *, to=PUBLIC, by=None)

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

Parameters:

• ids (list[str])

• to (str)

• by (str | None)

Return type:

list[str]

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

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

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

Parameters:

• team (str)

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

Return type:

set[str]

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

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)

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)

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)

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)

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

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

Parameters:

• substrate (Substrate)

• scope (str | None)

Return type:

dict[str, Any]

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

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)

• visited (int)

as_dict()

Return a JSON-serializable lineage report.

Return type:

dict[str, Any]

class Governance(approvers=<factory>)

Bases: object

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

Parameters:

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

may_approve(who, scope)

Return whether who is allowed to approve promotion into scope.

Parameters:

• who (str)

• scope (str)

Return type:

bool

grant(who, scope)

Add who as an approver for scope (chainable).

Parameters:

• who (str)

• scope (str)

Return type:

Governance

propose(substrate, ids, *, to, by=None)

Mark items as pending promotion to scope to; they are not yet visible there. Returns the ids.

Parameters:

• substrate (Substrate)

• ids (list[str])

• to (str)

• by (str | None)

Return type:

list[str]

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

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

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)

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

Parameters:

• substrate (Substrate)

• to (str | None)

Return type:

list[SubstrateItem]

detect_secrets(text)

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)

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:

• text (str)

• mask (str)

• keep_prefix (int)

Return type:

str

safe_text(text)

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

Parameters:

text (str)

Return type:

str

scan_item(item)

Scan a substrate item’s text surface for secrets.

Parameters:

item (Any)

Return type:

SecretScan

scan_substrate(substrate, *, scope=None)

Sweep a substrate for leaked secrets and report which stored items triggered rules.

Returns {n_items, n_dirty, dirty: [{item_id, rules}]} for compatibility with existing callers; entries in dirty are the items that matched one or more secret-detection rules.

Parameters:

• substrate (Any)

• scope (str | None)

Return type:

dict[str, Any]

class SecretScan(findings=<factory>)

Bases: object

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

Parameters:

findings (list[SecretFinding])

property clean: bool

Whether the scan found no secrets.

rules()

Return the sorted names of triggered secret-detection rules.

Return type:

list[str]

as_dict()

Return a JSON-serializable scan summary.

Return type:

dict[str, Any]

class SecretFinding(rule, start, end, preview)

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)

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

Semantic-entropy cutoff used to decide whether answers are trusted.

answer(prompt)

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

Parameters:

prompt (Any)

Return type:

ExternalAnswer

confident(prompt)

Return whether the external model is calibrated-confident on prompt.

Parameters:

prompt (Any)

Return type:

bool

class ExternalAnswer(prompt, answer, entropy, confident)

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)

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)

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:

• triples (Any)

• ontology (Any)

• types (dict[str, str] | None)

• name (str)

• cost (float)

• description (str)

• k (int)

Return type:

Any

link_entities(question, entities)

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)

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 an unvalidated store cannot inject a type-invalid fact as evidence.

Parameters:

• triples (Any)

• question (str)

• ontology (Any)

• types (dict[str, str] | None)

• k (int)

Return type:

dict[str, Any]

check_freshness(substrate, item_id, *, max_age_s=None, now=None)

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)

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)

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)

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)

as_dict()

Return a JSON-serializable freshness verdict.

Return type:

dict[str, Any]

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

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)

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)

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)

as_dict()

Return a JSON-serializable harness result.

Return type:

dict[str, Any]

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

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:

• substrate (Substrate)

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

• escalate (Callable[[str, Any], str] | None)

• max_chars (int)

Return type:

Harness

monitoring_harness(reasoner, *, escalate=None)

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

Parameters:

• reasoner (Reasoner)

• escalate (Callable[[str, Any], str] | None)

Return type:

Harness

register_harness(substrate, harness, *, scope='local')

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)

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

Parameters:

• substrate (Substrate)

• query (str)

• scope (str | None)

Return type:

list[dict[str, Any]]

harvest_knowledge(model_output, *, source, extract=None)

Split model_output into atomic claims, each stamped with source (which model produced it, its confidence, etc). Default extraction is sentence-level (mixle.reason.llm.sentence_claims()); pass extract for a different atomic-proposition splitter.

Parameters:

• model_output (str)

• source (dict[str, Any])

• extract (Callable[[str], list[str]] | None)

Return type:

list[Claim]

assimilate(sub, claim, evidence, *, scope='local')

Bayesian-ish update of the belief in claim from evidence – never a binary write.

Finds or creates the belief item (keyed on normalized claim text) and appends each evidence entry ({"source_id", "tier", "direction": "+"/"-", "weight"}); tier must be one of _TIER_STRENGTH ("model_assertion" plus mixle.doe.oracle.VERIFIABILITY_TIERS).

Anti-laundering: an entry whose source_id resolves back to THIS belief (a cycle) or to another belief item with no independent (non-model-assertion) support of its own is stored with an effective weight of zero – it cannot move the credence, though it stays in the trail for audit.

Parameters:

• sub (Substrate)

• claim (Claim)

• evidence (dict[str, Any] | Sequence[dict[str, Any]])

• scope (str)

Return type:

BeliefItem

retract(sub, source_id, *, scope=None)

Remove every evidence entry citing source_id, recomputing credence, and CASCADE: if that removal causes a belief to lose its only independent support, also strip citations of THAT belief from whatever cited it, recursively. Returns every belief item touched by the cascade.

Parameters:

• sub (Substrate)

• source_id (str)

• scope (str | None)

Return type:

list[BeliefItem]

retrieve_beliefs(sub, query, *, k=8, min_credence=None, scope=None)

Beliefs relevant to query, optionally thresholded on min_credence and re-ranked by relevance * credence – so a caller can weight by, or hard-filter on, how much the store actually believes each item (never a “fact” vs “non-fact” partition, only credence).

Parameters:

• sub (Substrate)

• query (str)

• k (int)

• min_credence (float | None)

• scope (str | None)

Return type:

list[BeliefItem]

credence_from_history(evidence_history)

The credence implied by an evidence history alone – a pure function, so replaying a belief’s stored evidence_history through this always reproduces its current credence exactly.

Parameters:

evidence_history (Sequence[EvidenceEntry])

Return type:

float

class BeliefItem(id, claim, credence, evidence_history=<factory>, scope='local')

Bases: object

A claim’s current credence plus the full evidence trail that produced it.

Parameters:

• id (str)

• claim (Claim)

• credence (float)

• evidence_history (list[EvidenceEntry])

• scope (str)

class Claim(text, produced_by=<factory>, quantity=None)

Bases: object

One atomic proposition (or typed quantity) pulled from a model’s output.

Parameters:

• text (str)

• produced_by (dict[str, Any])

• quantity (float | None)

class EvidenceEntry(source_id, tier, direction='+', weight=1.0, time=<factory>)

Bases: object

One piece of evidence that moved a belief’s credence, in the order it was applied.

Parameters:

• source_id (str)

• tier (str)

• direction (str)

• weight (float)

• time (float)

Submodules

• mixle.substrate.accum module
• mixle.substrate.act module
• mixle.substrate.answer module
• mixle.substrate.belief module
• mixle.substrate.context module
• mixle.substrate.core module
• mixle.substrate.eig_retrieve module
• mixle.substrate.factuality module
• mixle.substrate.freshness module
• mixle.substrate.governance module
• mixle.substrate.harness module
• mixle.substrate.ingest module
• mixle.substrate.interop module
• mixle.substrate.kg_rag module
• mixle.substrate.multihop module
• mixle.substrate.reasoner module
• mixle.substrate.retrieve module
• mixle.substrate.security module
• mixle.substrate.spaces module
• mixle.substrate.trust module