Reasoning engine

The reasoner is a pure, deterministic, LLM-free forward-chaining engine. Given base facts (from retrieved hyperedges) and a set of rules, it derives new facts to a fixpoint, stratum by stratum, honouring negation-as-failure. Termination is guaranteed: there are no function symbols, the Herbrand base is finite, and derivation is monotone.

RetrievedEdge[]  →  facts_from_edges  →  FactBase  →  SymbolicReasoner.run  →  ReasonResult

Facts

A Fact is a ground atom — a predicate with constant arguments. Its identity is (pred, args), so it is hashable and de-duplicated.

from hypermesh.rag.symbolic import Fact

f = Fact("coordinated_threat", (42,))
str(f)        # "coordinated_threat(42)"
f.pred        # "coordinated_threat"
f.args        # (42,)

Each fact in a FactBase carries a FactMeta:

Field	Meaning
source	"base" (from a hyperedge) or "derived" (produced by a rule).
confidence	[0, 1].
support	Premise facts a derived fact was produced from (tuple[Fact, ...]).
rule_id	The rule that produced a derived fact.
edge_ref	For base facts: the originating edge {table, edge_idx, event_ts, member_ids, hedge_tag}.

The edge → fact bridge

facts_from_edges(edges, table="") projects retrieved hyperedges into a base FactBase. This is the seam that keeps reasoning in continuity with your ingested hypergraph — the reasoner never sees anything that isn’t grounded in a stored edge.

from hypermesh.rag.symbolic import facts_from_edges

fb = facts_from_edges(edges, table="DRONES")
len(fb)                       # number of base facts
fb.by_pred("formation")       # [Fact("formation", (E, "CONVERGENCE")), ...]

For each edge E it emits hyperedge(E), formation(E, F), weight(E, W), at(E, Ts), and per member member(E, A) + member_type(A, T). Every base fact records its edge_ref (including a HEDGE-N tag) for later proof binding.

FactBase

from hypermesh.rag.symbolic import FactBase, Fact

fb = FactBase()
fb.add_base(Fact("member", (1, 7)), confidence=1.0, edge_ref={...})   # -> bool (newly added?)
fb.add_derived(Fact("threat", (1,)), support=[...], rule_id="r1", confidence=0.9)

fb.by_pred("threat")   # facts for one predicate (indexed, fast)
fb.meta(fact)          # FactMeta for a fact
fact in fb             # membership
len(fb); list(fb)      # size / iterate all facts
fb.derived()           # only derived facts

add_base / add_derived return True only when the fact is newly added. Re-adding an existing fact is a no-op — this monotonicity is what guarantees the fixpoint loop terminates.

Stratification

stratify(rules) assigns each predicate an integer stratum so that a predicate appearing under not is computed in a strictly lower stratum than the head that negates it. Base predicates are stratum 0.

from hypermesh.rag.symbolic import stratify

strata = stratify(rules)   # -> {pred: int}, or raises StratificationError

The algorithm relaxes two constraints to a fixpoint over len(preds)+1 passes:

• stratum[head] ≥ stratum[q] for each positive body predicate q;
• stratum[head] > stratum[q] for each negated body predicate q.

If it cannot converge — i.e. negation participates in a cycle — it raises StratificationError. A final pass re-checks every strict (negative) constraint to be certain. This is what makes negation-as-failure sound and the result unique.

Body matching

match_body(rule, fb) yields every (binding, support_facts) for which the rule body is satisfied: it joins the positive atoms (unifying variables), filters by all comparisons, then drops any binding whose negated atoms are present. It’s exposed for testing and custom tooling.

from hypermesh.rag.symbolic import match_body

for binding, support in match_body(rule, fb):
    print(binding)     # {"?E": 42, "?A": 7, "?W": 0.91}
    print(support)     # [Fact("member", (42, 7)), Fact("weight", (42, 0.91)), ...]

SymbolicReasoner

from hypermesh.rag.symbolic import SymbolicReasoner

reasoner = SymbolicReasoner(max_iterations=64, deadline_ms=None)
result = reasoner.run(fact_base, rules)

run first stratifies the enabled rules, groups them by their head’s stratum, and sorts each group by (-priority, id) for deterministic firing order. It then evaluates strata in ascending order, looping each stratum to a fixpoint. Derived facts are written back into the same FactBase (so proofs can be built from it).

Bounds. max_iterations caps the per-stratum fixpoint loop; deadline_ms (optional) sets a wall-clock budget. If either is hit, reasoning stops early and ReasonResult.timed_out is True — partial results remain sound (everything derived is still proven), just possibly incomplete.

ReasonResult

Field	Meaning
fact_base	The FactBase, now containing derived facts.
derived	list[Fact] of newly derived facts, in derivation order.
rules_fired	Ordered list of rule ids that produced at least one fact.
iterations	Total fixpoint iterations across all strata.
strata	{pred: int} stratum assignment.
timed_out	True if a bound stopped reasoning early.

Determinism & termination

• Deterministic — same facts + same rules ⇒ same derived set, every time. Output ordering is stabilised by the pipeline before exposure.
• Terminating — finite Herbrand base + monotone derivation ⇒ guaranteed fixpoint, with iteration/time bounds as belt-and-braces.
• Sound — every derived fact is justified by a concrete rule firing over facts that trace to stored hyperedges.

Next: Proofs & firewall.