📋
jugeo.geometry

Geometry

Core building blocks of a semantic site: coordinates, morphisms, covering families, and the site builder.

enum CoordinateKind

Enumeration of semantic positions a coordinate may occupy within a Python codebase.

NameValueDescription
MODULE"module"Top-level Python module
PACKAGE"package"Python package (directory with __init__.py)
CLASS"class"Class body
FUNCTION"function"Function or method definition
STATEMENT"statement"Single statement (assignment, call, etc.)
EXPRESSION"expression"Sub-expression within a statement
BLOCK"block"Arbitrary named region inside a function
TYPE"type"Type annotation position
dataclass Coordinate Coordinate(kind, path, name, parent=None, meta=None)

A point in the semantic site — the smallest addressable unit of a Python program that JuGeo can reason about. Coordinates are immutable and hashable, serving as keys in descent data structures.

ParameterTypeDescriptionDefault
kindCoordinateKindThe semantic kind of this coordinaterequired
pathstrDotted module path, e.g. "mylib.core.util"required
namestrLocal name within the containing scoperequired
parentCoordinate | NoneEnclosing coordinate, or None for top-levelNone
metadict | NoneArbitrary metadata (line numbers, AST node refs, etc.)None
  • fqn() -> strFully-qualified name: path + "." + name.
  • is_ancestor_of(other: Coordinate) -> boolTrue if self is a proper ancestor in the containment hierarchy.
  • children(site: Site) -> list[Coordinate]Return all direct children registered on the given site.
from jugeo.geometry import Coordinate, CoordinateKind

mod = Coordinate(("mylib", "core"), CoordinateKind.MODULE)
fn = Coordinate(("mylib", "core", "parse"), CoordinateKind.FUNCTION, metadata={"lineno": 42})

print(fn.components)
print(fn.name)
print(fn.metadata["lineno"])
class Site Site(name, coordinates, morphisms, topology)

A Grothendieck site over a Python codebase. Packages together a category of coordinates and morphisms with a Grothendieck topology (encoded as covering families). Normally created via build_site() or SiteBuilder.

ParameterTypeDescriptionDefault
namestrHuman-readable name for the siterequired
coordinateslist[Coordinate]All coordinates in the categoryrequired
morphismslist[Morphism]All morphisms between coordinatesrequired
topologylist[CoveringFamily]Covering families defining the topologyrequired
  • covering_sieves(c: Coordinate) -> list[CoveringFamily]All covering families whose target is c.
  • hom(src: Coordinate, tgt: Coordinate) -> list[Morphism]Morphisms from src to tgt.
  • sheaf_condition(F, c) -> boolCheck that presheaf F satisfies the sheaf condition at c.
  • restrict(J: Judgment, m: Morphism) -> JudgmentPull back a judgment along a morphism.
class SiteBuilder SiteBuilder(name)

Fluent builder for constructing Site objects incrementally. Validates consistency (e.g., all morphism endpoints exist) before finalising.

ParameterTypeDescriptionDefault
namestrSite name propagated to the resulting Siterequired
  • add_coordinate(c: Coordinate) -> SiteBuilderRegister a coordinate; returns self for chaining.
  • add_morphism(m: Morphism) -> SiteBuilderRegister a morphism; raises ValueError if endpoints are unknown.
  • add_covering(cf: CoveringFamily) -> SiteBuilderRegister a covering family.
  • build() -> SiteValidate and construct the Site. Raises SiteError on inconsistency.
from jugeo.geometry import SiteBuilder, Coordinate, CoordinateKind, Morphism
from jugeo.geometry.site import MorphismKind

builder = SiteBuilder("my_project")
mod = Coordinate(("mylib",), CoordinateKind.MODULE)
fn = Coordinate(("mylib", "parse"), CoordinateKind.FUNCTION)
builder.add_coordinate(mod).add_coordinate(fn)
builder.add_morphism(Morphism(fn, mod, MorphismKind.RESTRICTION, "contained-in"))
site = builder.build()
print(site.coordinate_count(), site.morphism_count())
dataclass Morphism Morphism(source, target, kind, label=None)

A directed arrow between two coordinates representing a structural or semantic relationship. Common kind values are "inclusion", "call", "import", "subtype", and "data_flow".

ParameterTypeDescriptionDefault
sourceCoordinateDomain of the morphismrequired
targetCoordinateCodomain of the morphismrequired
kindstrSemantic type of the relationshiprequired
labelstr | NoneOptional human-readable labelNone
dataclass CoveringFamily CoveringFamily(target, members, name=None)

A sieve over a coordinate: a collection of morphisms whose sources jointly cover the target, satisfying the Grothendieck topology axioms (stability and transitivity).

ParameterTypeDescriptionDefault
targetCoordinateThe coordinate being coveredrequired
memberslist[Morphism]Morphisms that form the covering sieverequired
namestr | NoneOptional name (used in obstruction messages)None
function build_site build_site(root_path, *, depth=None, kinds=None, include_calls=True) -> Site

Automatically construct a Site by introspecting a Python package on disk. Parses ASTs, extracts coordinates at each requested CoordinateKind, and infers morphisms from containment and import relationships.

ParameterTypeDescriptionDefault
root_pathstr | PathFilesystem path to the package rootrequired
depthint | NoneMaximum containment depth; None means unlimitedNone
kindsset[CoordinateKind] | NoneCoordinate kinds to include; None includes allNone
include_callsboolWhether to add call-graph morphismsTrue
from jugeo.geometry import build_site, Coordinate, CoordinateKind

mod = Coordinate(("mylib",), CoordinateKind.MODULE)
fn = Coordinate(("mylib", "parse"), CoordinateKind.FUNCTION)
site = build_site([mod, fn])
print(site.coordinate_count(), site.morphism_count())
function restrict_coordinate restrict_coordinate(site, c, morphism) -> Coordinate

Pull back a coordinate along a morphism within the site. Equivalent to computing the fiber product in the category of coordinates.

ParameterTypeDescriptionDefault
siteSiteThe ambient siterequired
cCoordinateSource coordinate to be restrictedrequired
morphismMorphismThe morphism along which to pull backrequired
jugeo.geometry.descent

Descent

Engine for local-to-global gluing of judgments, obstruction detection, and cohomological repair strategies.

enum DescentStrategy

Strategy governing how the DescentEngine handles incompatible local sections.

NameDescription
STRICTFail immediately on any gluing mismatch
PERMISSIVECollect all obstructions and continue
REPAIRAttempt automatic repair via RepairFrontier before failing
ORACLEDelegate mismatches to an external oracle channel
class DescentEngine DescentEngine(site, strategy=DescentStrategy.PERMISSIVE, repair_frontier=None)

Orchestrates the descent process: given a collection of LocalSections over a covering family, attempts to glue them into a GlobalSection. Detects and classifies DescentObstructions when gluing fails.

ParameterTypeDescriptionDefault
siteSiteThe ambient semantic siterequired
strategyDescentStrategyHow to handle gluing failuresPERMISSIVE
repair_frontierRepairFrontier | NoneRepair oracle used when strategy is REPAIRNone
  • glue(sections: list[LocalSection], cover: CoveringFamily) -> GlobalSection | DescentObstructionMain entry point. Attempts to glue local sections over the given cover.
  • check_cocycle(data: GluingData) -> list[DescentObstruction]Verify the cocycle condition on overlaps. Returns empty list on success.
  • obstruction_class(obs: DescentObstruction) -> CohomologyClassCompute the first Cech cohomology class witnessing the obstruction.
  • repair(obs: DescentObstruction) -> GlobalSection | NoneAttempt repair; returns None if repair is impossible.
from jugeo.geometry.descent import DescentEngine, DescentStrategy

engine = DescentEngine()
print(type(engine).__name__)
print([strategy.name for strategy in DescentStrategy])
print(callable(engine.attempt_descent))
dataclass LocalSection LocalSection(coordinate, judgment, morphism)

A judgment localised to a single member of a covering family. The morphism connects the local coordinate back to the covered object.

ParameterTypeDescriptionDefault
coordinateCoordinateThe local coordinate (cover member)required
judgmentJudgmentThe judgment at this local patchrequired
morphismMorphismThe inclusion morphism into the covered coordinaterequired
dataclass GlobalSection GlobalSection(coordinate, judgment, provenance)

A successfully glued global judgment. The provenance field records which local sections were combined and through which covering family.

ParameterTypeDescriptionDefault
coordinateCoordinateThe global coordinaterequired
judgmentJudgmentThe assembled global judgmentrequired
provenanceGluingDataRecord of the gluing that produced this sectionrequired
dataclass DescentObstruction DescentObstruction(kind, coordinate, conflicting, cohomology_class, message)

Records a failure of the gluing/descent process. The cohomology_class encodes the algebraic obstruction in \(\check{H}^1\).

ParameterTypeDescriptionDefault
kindstrOne of "cocycle", "trust", "evidence", "formula"required
coordinateCoordinateWhere the obstruction was detectedrequired
conflictinglist[LocalSection]The incompatible local sectionsrequired
cohomology_classCohomologyClassThe \(\check{H}^1\) class witnessing the obstructionrequired
messagestrHuman-readable explanationrequired
dataclass GluingData GluingData(cover, sections, overlaps, cocycle_maps)

Complete record of a gluing attempt: the covering family, local sections, pairwise overlap data, and the cocycle maps used to check compatibility.

dataclass CohomologyClass CohomologyClass(degree, representative, is_trivial)

An element of \(\check{H}^n\) for the semantic site. Degree 0 is sections; degree 1 encodes gluing obstructions. is_trivial is True when the class vanishes (successful gluing).

class RepairFrontier RepairFrontier(strategies, budget=None)

A priority queue of repair strategies applied when the DescentEngine encounters an obstruction. Each strategy is tried in priority order until one succeeds or the budget is exhausted.

ParameterTypeDescriptionDefault
strategieslist[Callable]Ordered list of repair functions (obs) -> LocalSection | Nonerequired
budgetint | NoneMaximum repair attempts; None is unlimitedNone
  • attempt(obs: DescentObstruction) -> LocalSection | NoneTry each strategy in turn; return the repaired section or None.
  • add_strategy(fn, priority=0) -> NoneRegister a new repair strategy with the given priority.
jugeo.geometry.covers

Covers

Builder and data structures for constructing covering families and inspecting pairwise overlaps.

dataclass Cover Cover(target, members, name=None)

A validated covering family. Unlike raw CoveringFamily, a Cover has been checked for the stability and transitivity axioms of the Grothendieck topology.

ParameterTypeDescriptionDefault
targetCoordinateCoordinate being coveredrequired
memberslist[CoverMember]Ordered list of cover membersrequired
namestr | NoneDescriptive nameNone
  • overlaps() -> list[OverlapDatum]Compute all pairwise intersections of cover members.
  • is_good_cover() -> boolTrue if all pairwise and higher overlaps are contractible.
  • refine(extra: list[CoverMember]) -> CoverReturn a refined cover by adding extra members.
dataclass CoverMember CoverMember(coordinate, inclusion, weight=1.0)

A single patch in a covering family. The weight is used by budget-allocation strategies to prioritise verification effort.

ParameterTypeDescriptionDefault
coordinateCoordinateThe local patch coordinaterequired
inclusionMorphismInclusion into the covered targetrequired
weightfloatRelative verification effort weight1.0
class CoverBuilder CoverBuilder(target, site)

Fluent builder for Cover objects. Validates topology axioms on build().

ParameterTypeDescriptionDefault
targetCoordinateThe coordinate to be coveredrequired
siteSiteAmbient site for morphism validationrequired
  • add_member(c: Coordinate, inclusion: Morphism, weight=1.0) -> CoverBuilderAdd a patch to the cover.
  • build() -> CoverValidate and construct the Cover.
from jugeo.geometry import Coordinate, CoordinateKind, Morphism
from jugeo.geometry.site import MorphismKind
from jugeo.geometry.covers import CoverBuilder

base = Coordinate(("pricing",), CoordinateKind.MODULE)
patch = Coordinate(("pricing", "total"), CoordinateKind.FUNCTION)
inclusion = Morphism(patch, base, MorphismKind.RESTRICTION, "contained-in")
cover = CoverBuilder().set_base(base).add_member(patch, inclusion).build()
print(len(cover.members))
print(type(cover).__name__)
dataclass OverlapDatum OverlapDatum(left, right, intersection, left_restriction, right_restriction)

Records the intersection of two cover members together with the restriction morphisms from each member into the intersection. Used by DescentEngine.check_cocycle().

ParameterTypeDescriptionDefault
leftCoverMemberFirst patchrequired
rightCoverMemberSecond patchrequired
intersectionCoordinateThe pairwise overlap coordinaterequired
left_restrictionMorphismRestriction from left patch to intersectionrequired
right_restrictionMorphismRestriction from right patch to intersectionrequired
jugeo.judgments

Judgments

The 4-tuple judgment type and all component classes: semantic topos, proof structure, cohomological spectrum, and graded evaluation.

dataclass Judgment Judgment(coordinate, proposition, carrier, evidence, obligation, obstruction, trust, proof_object=None)

The fundamental unit of reasoning in JuGeo. A judgment \(J = (c,\varphi,A,E,O,B,T,\Pi)\) ties a logical claim to a specific coordinate, its supporting evidence, any residual proof obligations, and a trust annotation.

ParameterTypeDescriptionDefault
coordinateCoordinate\(c\) — where this judgment livesrequired
propositionProposition\(\varphi\) — the logical claimrequired
carrierCarrier\(A\) — antecedent judgments this one depends onrequired
evidenceEvidenceBundle\(E\) — multi-channel evidence supporting the claimrequired
obligationResidualObligation\(O\) — proof obligations not yet dischargedrequired
obstructionObstruction\(B\) — known obstructions to discharging Orequired
trustTrustAnnotation\(T\) — trust level and provenance metadatarequired
proof_objectobject | None\(\Pi\) — optional proof term (Lean export, Z3 model, etc.)None
  • restrict(m: Morphism) -> JudgmentPull back along a morphism; propagates evidence and obligations.
  • compose(other: Judgment) -> JudgmentSequential composition: self feeds into other as antecedent.
  • is_closed() -> boolTrue if all residual obligations are discharged.
  • trust_level() -> TrustLevelShortcut for self.trust.level.
  • to_dict() -> dictSerialise to a JSON-compatible dictionary.
from jugeo import judgments

print(sorted(name for name in dir(judgments) if not name.startswith("_")))
dataclass Proposition Proposition(formula, language="jugeo", free_vars=None)

A logical formula. The language field indicates the syntax: "jugeo" for native JuGeo logic, "z3smt" for SMT-LIB2, "lean4" for Lean 4 term syntax.

ParameterTypeDescriptionDefault
formulastrThe formula textrequired
languagestrSyntax dialect: "jugeo", "z3smt", or "lean4""jugeo"
free_varslist[str] | NoneNames of free variables occurring in the formulaNone
dataclass Carrier Carrier(antecedents, kind="conjunction")

The set of antecedent judgments on which a judgment depends. kind specifies how antecedents combine: "conjunction" (all required) or "disjunction" (any sufficient).

ParameterTypeDescriptionDefault
antecedentslist[Judgment]Prior judgments this one depends onrequired
kindstr"conjunction" or "disjunction""conjunction"
dataclass EvidenceBundle EvidenceBundle(items, combination="weakest_link")

A collection of evidence items from potentially multiple channels. The overall trust level is derived by applying the combination rule across all items.

ParameterTypeDescriptionDefault
itemslist[EvidenceItem]Individual evidence contributionsrequired
combinationstr"weakest_link", "strongest", or "weighted""weakest_link"
  • effective_trust() -> TrustLevelAggregate trust level according to the combination rule.
  • channel_summary() -> dict[str, int]Count items per channel name.
dataclass EvidenceItem EvidenceItem(channel, payload, trust_level, timestamp=None)

A single piece of evidence from one channel. payload is channel-specific (e.g., an SMT proof certificate, a Lean term, a test result).

ParameterTypeDescriptionDefault
channelstrChannel name, e.g. "z3", "lean", "pytest"required
payloadobjectChannel-specific proof artifactrequired
trust_levelTrustLevelTrust level awarded to this evidence itemrequired
timestampfloat | NoneUnix timestamp of evidence productionNone
dataclass ResidualObligation ResidualObligation(goals, priority=0)

The set of open proof goals remaining after current evidence has been applied. An empty goals list means the judgment is fully discharged.

ParameterTypeDescriptionDefault
goalslist[Proposition]Open sub-goalsrequired
priorityintScheduling priority for the solver router0
  • is_discharged() -> boolTrue iff goals is empty.
  • discharge(goal: Proposition) -> ResidualObligationReturn a new obligation with goal removed.
dataclass Obstruction Obstruction(kind, description, cohomology_class=None)

A known impediment to discharging the residual obligation. Use Obstruction.none() for judgments with no known obstruction.

ParameterTypeDescriptionDefault
kindstr | NoneCategory: "undecidable", "timeout", "trust_gap", etc.required
descriptionstrHuman-readable explanationrequired
cohomology_classCohomologyClass | NoneAlgebraic witness if availableNone
  • Obstruction.none() -> ObstructionCanonical sentinel representing the absence of any obstruction.
dataclass TrustAnnotation TrustAnnotation(level, source, profile=None)

Trust metadata attached to a judgment. Records the enumerated trust level and the identity of the agent or tool that produced the evidence.

ParameterTypeDescriptionDefault
levelTrustLevelEnumerated trust levelrequired
sourcestrIdentifier of the evidence producer (tool name, agent ID, etc.)required
profileTrustProfile | NoneFull trust profile, if availableNone
jugeo.evidence.trust

Trust

The seven-level trust lattice, trust profiles, and the algebraic operations that combine and compare trust values.

enum TrustLevel

Ordered enumeration of trust levels, from highest to lowest confidence. Levels form a lattice with VERIFIED at the top and CONTRADICTED as a distinguished bottom element indicating active refutation.

NameRankDescription
VERIFIED6Machine-checked formal proof (e.g., Lean 4)
SOLVER5SMT/SAT solver decision procedure
RUNTIME4Property observed at runtime (tests, monitoring)
ORACLE3Trusted external oracle or human review
COPILOT2LLM/Copilot suggestion, not independently verified
UNVERIFIED1Asserted without supporting evidence
CONTRADICTED0Active counter-evidence exists; claim is refuted
from jugeo.evidence.trust import TrustLevel, TrustAlgebra

algebra = TrustAlgebra()
print([level.name for level in TrustLevel])
print(algebra.join(TrustLevel.RUNTIME_WITNESSED, TrustLevel.SOLVER_DISCHARGED))
dataclass TrustProfile TrustProfile(level, channels, decay_rate=0.0, confidence=1.0)

A richer trust descriptor recording not just the level but the contributing channels, a temporal decay rate (for runtime evidence), and an overall confidence score in [0, 1].

ParameterTypeDescriptionDefault
levelTrustLevelBase trust levelrequired
channelslist[str]Evidence channels contributing to this profilerequired
decay_ratefloatConfidence loss per day (0 = no decay)0.0
confidencefloatScalar confidence in [0.0, 1.0]1.0
  • age(days: float) -> TrustProfileReturn a new profile with confidence decayed by days * decay_rate.
  • effective_level(threshold=0.5) -> TrustLevelDowngrade the level if confidence falls below threshold.
class TrustAlgebra TrustAlgebra()

Stateless class providing lattice operations over TrustLevel and TrustProfile. All methods are classmethods.

  • meet(a: TrustLevel, b: TrustLevel) -> TrustLevelGreatest lower bound (weakest-link combination).
  • join(a: TrustLevel, b: TrustLevel) -> TrustLevelLeast upper bound (strongest-wins combination).
  • combine(profiles: list[TrustProfile], rule="meet") -> TrustProfileCombine multiple profiles using "meet", "join", or "weighted".
  • transfer(J: Judgment, m: Morphism) -> TrustAnnotationCompute the trust annotation for a judgment restricted along m.
  • sufficient_for(level: TrustLevel, requirement: TrustLevel) -> boolTrue if level >= requirement in the lattice order.
from jugeo.evidence.trust import TrustAlgebra, TrustLevel

ta = TrustAlgebra()
print(ta.join(TrustLevel.RUNTIME_WITNESSED, TrustLevel.SOLVER_DISCHARGED))
print(ta.promote(TrustLevel.COPILOT_SUGGESTED, "solver witness"))
jugeo.solver

Solver

SMT routing, Z3 session management, formula fragments, and routing strategies for dispatching verification queries.

enum FragmentTag

Identifies the logical fragment of an SMT formula. Used by SolverRouter to select the most efficient solver tactic.

NameDescription
LIALinear Integer Arithmetic
LRALinear Real Arithmetic
NIANonlinear Integer Arithmetic
BVBit-vectors
ARRAYTheory of arrays
UFUninterpreted functions
HORNConstrained Horn Clauses
MIXEDMulti-theory formula requiring combination
enum RoutingStrategy

High-level policy for the SolverRouter when multiple solver sessions are available.

NameDescription
FRAGMENT_FIRSTSelect solver by fragment tag; fall back to general solver
PORTFOLIORun all compatible solvers in parallel; take first result
SEQUENTIALTry solvers in priority order until one succeeds
BUDGET_AWARESelect solver to minimise expected cost given a time budget
dataclass Z3Formula Z3Formula(smt2_text, fragment, timeout_ms=5000, label=None)

A formula ready for dispatch to Z3, expressed in SMT-LIB2 syntax. fragment is pre-classified to allow the router to pick the right tactic.

ParameterTypeDescriptionDefault
smt2_textstrSMT-LIB2 formula text including (check-sat)required
fragmentFragmentTagPre-classified logical fragmentrequired
timeout_msintPer-query timeout in milliseconds5000
labelstr | NoneHuman-readable label for loggingNone
dataclass Z3Result Z3Result(status, model, proof, elapsed_ms, formula)

Result from a Z3 query. status is one of "sat", "unsat", or "unknown". model is populated on "sat"; proof on "unsat" when proof generation is enabled.

  • is_proved() -> boolTrue iff status is "unsat" (negation unsatisfiable means original formula is valid).
  • to_evidence_item() -> EvidenceItemConvert to an EvidenceItem with trust SOLVER (unsat) or RUNTIME (sat-model).
class Z3Session Z3Session(config=None)

Manages a persistent Z3 context, supporting incremental solving (push/pop), tactic selection, and result caching. Thread-safe when config.threadsafe=True.

  • solve(formula: Z3Formula) -> Z3ResultSubmit a formula and return the result.
  • solve_batch(formulas: list[Z3Formula]) -> list[Z3Result]Solve multiple formulas, sharing the Z3 context.
  • push() / pop()Incremental checkpoint operations.
  • reset()Clear all assertions and restore the initial context.
from jugeo.solver.z3_session import Z3Session, Z3Formula, FragmentTag

session = Z3Session()
formula = Z3Formula(FragmentTag.QF_LIA, "(assert (> x 0))")
print(type(session).__name__)
print(formula.kind)
print(formula.expression)
class SolverRouter SolverRouter(sessions, strategy=RoutingStrategy.FRAGMENT_FIRST, budget_ms=None)

Routes verification queries to the appropriate Z3 session or solver backend based on formula fragment, strategy, and available budget.

ParameterTypeDescriptionDefault
sessionslist[Z3Session]Pool of solver sessions for dispatchrequired
strategyRoutingStrategyDispatch policyFRAGMENT_FIRST
budget_msint | NoneTotal wall-clock budget across all queries; None is unlimitedNone
  • dispatch(formula: Z3Formula) -> Z3ResultRoute a single formula to the best session.
  • dispatch_judgment(J: Judgment) -> JudgmentDispatch all open obligations in J; return an updated judgment.
  • stats() -> dictReturn routing statistics: calls per session, success rates, total time.
from jugeo.solver.router import SolverRouter, RoutingStrategyKind

router = SolverRouter()
print(type(router).__name__)
print([strategy.name for strategy in RoutingStrategyKind][:4])
jugeo.evidence.channels

Evidence Channels

Pluggable evidence channels, routing, and federation for combining evidence from heterogeneous sources.

class EvidenceChannel EvidenceChannel(name, trust_level, timeout_ms=10000)

Abstract base class for all evidence channels. Concrete subclasses implement query() to obtain evidence for a proposition at a coordinate. Built-in channels: Z3Channel, LeanChannel, PytestChannel, OracleChannel, CopilotChannel.

ParameterTypeDescriptionDefault
namestrUnique channel identifierrequired
trust_levelTrustLevelMaximum trust level this channel can awardrequired
timeout_msintPer-query timeout in milliseconds10000
  • query(proposition: Proposition, coordinate: Coordinate) -> EvidenceItem | NoneAttempt to produce evidence. Returns None on failure or timeout.
  • is_available() -> boolCheck whether the backing tool or service is reachable.
from jugeo.evidence.channels import EvidenceChannel

print([channel.name for channel in EvidenceChannel])
print(EvidenceChannel.COPILOT.requires_corroboration())
class ChannelRouter ChannelRouter(channels, policy="parallel", min_trust=None)

Routes a proposition to one or more evidence channels according to a dispatching policy and aggregates results into an EvidenceBundle.

ParameterTypeDescriptionDefault
channelslist[EvidenceChannel]Available channels in priority orderrequired
policystr"parallel" (query all), "cascade" (stop on first success), "required_all""parallel"
min_trustTrustLevel | NoneAccept only evidence at or above this levelNone
  • gather(proposition: Proposition, coordinate: Coordinate) -> EvidenceBundleQuery channels per policy and aggregate results.
  • add_channel(ch: EvidenceChannel, priority=0) -> NoneRegister a new channel at the given priority.
from jugeo.evidence.channels import ChannelRouter

router = ChannelRouter()
print(type(router).__name__)
print([name for name in dir(router) if not name.startswith("_")][:5])
class ChannelFederation ChannelFederation(routers, merge="union")

Federates multiple ChannelRouter instances, each potentially representing a separate node in a distributed verification network. Evidence from all routers is merged into a single EvidenceBundle.

ParameterTypeDescriptionDefault
routerslist[ChannelRouter]The federated routersrequired
mergestr"union" (collect all items) or "consensus" (require agreement)"union"
  • gather(proposition: Proposition, coordinate: Coordinate) -> EvidenceBundleFan out across all routers and merge results.
  • status() -> dict[str, bool]Availability status of each constituent router.
from jugeo.evidence.channels import ChannelFederation, ChannelRouter

federation = ChannelFederation(ChannelRouter())
print(type(federation).__name__)
print([name for name in dir(federation) if not name.startswith("_")][:4])