def solve(
self,
goals: list[Triple],
subst: Subst,
depth: int = 0,
allow_reorder: bool = True,
_visited: frozenset[Any] | None = None,
) -> Iterator[Subst]:
"""Prove goals with iterative DFS and trail-backed substitutions."""
goal_memo_key: str | None = None
goal_memo: dict[str, list[Subst]] | None = None
if depth == 0 and not _visited and not subst:
goal_memo = self._completed_goal_memo()
goal_memo_key = self._completed_goal_memo_key(goals, subst, allow_reorder)
cached = goal_memo.get(goal_memo_key)
if cached is not None:
for answer in cached:
yield dict(answer)
return
subst_mut: Subst = dict(subst)
trail: list[str] = []
visited_counts: dict[Any, int] = {key: 1 for key in (_visited or frozenset())}
visited_trail: list[Any] = []
answer_vars = set(subst)
def collect_vars(term: Term, target: set[str]) -> None:
if not self._term_needs_substitution(term):
return
if isinstance(term, Var):
target.add(term.name)
elif isinstance(term, ListTerm):
for item in term.elems:
collect_vars(item, target)
elif isinstance(term, OpenListTerm):
for item in term.prefix:
collect_vars(item, target)
target.add(term.tail_var)
elif isinstance(term, GraphTerm):
for triple in term.triples:
collect_vars(triple.s, target)
collect_vars(triple.p, target)
collect_vars(triple.o, target)
for original_goal in goals:
collect_vars(original_goal.s, answer_vars)
collect_vars(original_goal.p, answer_vars)
collect_vars(original_goal.o, answer_vars)
completed_answers: list[Subst] = []
def goal_key(goal: Triple) -> tuple[Any, Any, Any]:
# Standardized rule variables change names at every recursive
# call, so normalize variables for cycle detection. Blank nodes
# remain identity-bearing terms and must not be collapsed.
return (
self._visited_term_key(goal.s),
self._visited_term_key(goal.p),
self._visited_term_key(goal.o),
)
def undo_to(mark: int) -> None:
for name in reversed(trail[mark:]):
subst_mut.pop(name, None)
del trail[mark:]
def push_visited(key: Any) -> None:
visited_counts[key] = visited_counts.get(key, 0) + 1
visited_trail.append(key)
def undo_visited_to(mark: int) -> None:
for key in reversed(visited_trail[mark:]):
count = visited_counts.get(key, 0)
if count <= 1:
visited_counts.pop(key, None)
else:
visited_counts[key] = count - 1
del visited_trail[mark:]
def deref_trail(term: Term) -> Term:
if not isinstance(term, Var):
return term
while isinstance(term, Var):
value = subst_mut.get(term.name, _MISSING)
if value is _MISSING:
return term
term = value
return term
def apply_subst_trail(term: Term) -> Term:
if not self._term_needs_substitution(term):
return term
term = deref_trail(term)
if isinstance(term, ListTerm):
original = term.elems
size = len(original)
if size == 0:
return term
if size == 1:
e0 = apply_subst_trail(original[0])
return term if e0 == original[0] else ListTerm((e0,))
if size == 2:
e0 = apply_subst_trail(original[0])
e1 = apply_subst_trail(original[1])
return term if e0 == original[0] and e1 == original[1] else ListTerm((e0, e1))
if size == 3:
e0 = apply_subst_trail(original[0])
e1 = apply_subst_trail(original[1])
e2 = apply_subst_trail(original[2])
return (
term
if e0 == original[0] and e1 == original[1] and e2 == original[2]
else ListTerm((e0, e1, e2))
)
elems = tuple(apply_subst_trail(item) for item in original)
return term if elems == original else ListTerm(elems)
if isinstance(term, OpenListTerm):
prefix = tuple(apply_subst_trail(item) for item in term.prefix)
tail = apply_subst_trail(Var(term.tail_var))
if isinstance(tail, ListTerm):
return ListTerm((*prefix, *tail.elems))
if isinstance(tail, OpenListTerm):
return OpenListTerm((*prefix, *tail.prefix), tail.tail_var)
if isinstance(tail, Var):
if prefix == term.prefix and tail.name == term.tail_var:
return term
return OpenListTerm(prefix, tail.name)
return OpenListTerm(prefix, term.tail_var)
if isinstance(term, GraphTerm):
triples = tuple(apply_subst_triple_trail(triple) for triple in term.triples)
return term if triples == term.triples else GraphTerm(triples)
return term
def apply_subst_triple_trail(
triple: Triple,
ground_blanks: bool = False,
blank_mapping: dict[str, Blank] | None = None,
) -> Triple:
if ground_blanks:
return self._instantiate_head_triple(triple, subst_mut, blank_mapping)
s = apply_subst_trail(triple.s)
p = apply_subst_trail(triple.p)
o = apply_subst_trail(triple.o)
return triple if s == triple.s and p == triple.p and o == triple.o else Triple(s, p, o)
comparisons = {
"equalTo", "notEqualTo", "greaterThan", "lessThan",
"notGreaterThan", "notLessThan", "contains", "startsWith",
"endsWith", "matches", "notMatches", "notMember",
}
def unbound_trail(term: Term) -> int:
original = term
term = deref_trail(term)
if isinstance(original, Var) and isinstance(term, GraphTerm):
return 0
if isinstance(term, Var):
return 1
if isinstance(term, ListTerm):
elems = term.elems
size = len(elems)
if size == 0:
return 0
if size == 1:
return unbound_trail(elems[0])
if size == 2:
return unbound_trail(elems[0]) + unbound_trail(elems[1])
if size == 3:
return unbound_trail(elems[0]) + unbound_trail(elems[1]) + unbound_trail(elems[2])
return sum(unbound_trail(item) for item in elems)
if isinstance(term, GraphTerm):
return sum(
unbound_trail(triple.s) + unbound_trail(triple.p) + unbound_trail(triple.o)
for triple in term.triples
)
return 0
def goal_rank_trail(goal: Triple, pred: Term, handler: Callable[[BuiltinContext], list[Subst]] | None) -> tuple[int, int]:
subject_unbound = unbound_trail(goal.s)
object_unbound = unbound_trail(goal.o)
variables = subject_unbound + object_unbound
if not isinstance(pred, Iri):
return (0, variables)
if handler is None:
self._ensure_fact_indexes_current()
has_extensional_candidate = bool(self._facts_by_pred.get(self._lookup_key(pred)))
has_backward_rule = pred.value in self._backward_predicates
if has_backward_rule and not has_extensional_candidate and subject_unbound:
return (1, variables)
return (0, variables)
if pred.value == "http://www.w3.org/2000/10/swap/list#iterate" and subject_unbound == 0:
return (-1, variables)
if (
pred.value in {
"http://www.w3.org/2000/10/swap/list#append",
"http://www.w3.org/2000/10/swap/list#firstRest",
}
and object_unbound == 0
):
return (-1, variables)
local = pred.value.rsplit("#", 1)[-1]
if local in {"collectAllIn", "forAllIn"}:
return (1, variables)
if local in {"includes", "notIncludes"} and isinstance(deref_trail(goal.o), Var):
return (3, variables)
if local in {"includes", "notIncludes"} and variables:
return (1, variables)
if pred.value == LOG_NS + "equalTo":
left = deref_trail(goal.s)
right = deref_trail(goal.o)
if not (isinstance(left, Var) and isinstance(right, Var)):
return (-1, variables)
if local in comparisons and variables:
return (2, variables)
if subject_unbound == 0:
return (-1, variables)
return (2, variables)
def select_goal_index_trail(current_goals: list[Any]) -> int:
for index, goal in enumerate(current_goals):
if not isinstance(goal, Triple):
return index
predicate = deref_trail(goal.p)
handler = get_builtin(predicate.value) if isinstance(predicate, Iri) else None
rank = goal_rank_trail(goal, predicate, handler)
if handler is not None:
if rank[0] < 0:
return index
elif rank[0] == 0:
return index
return 0
def occurs(name: str, value: Term) -> bool:
value = deref_trail(value)
if not self._term_needs_substitution(value):
return False
if isinstance(value, Var):
return value.name == name
if isinstance(value, ListTerm):
elems = value.elems
size = len(elems)
if size == 0:
return False
if size == 1:
return occurs(name, elems[0])
if size == 2:
return occurs(name, elems[0]) or occurs(name, elems[1])
if size == 3:
return occurs(name, elems[0]) or occurs(name, elems[1]) or occurs(name, elems[2])
return any(occurs(name, item) for item in elems)
if isinstance(value, OpenListTerm):
return (
value.tail_var == name
or any(occurs(name, item) for item in value.prefix)
or occurs(name, Var(value.tail_var))
)
if isinstance(value, GraphTerm):
return any(
occurs(name, triple.s) or occurs(name, triple.p) or occurs(name, triple.o)
for triple in value.triples
)
return False
def bind_var(var: Var, value: Term) -> bool:
if var.name in subst_mut:
return unify_term_trail(subst_mut[var.name], value)
if isinstance(value, Var) and value.name == var.name:
return True
if occurs(var.name, value):
return False
subst_mut[var.name] = value
trail.append(var.name)
return True
def unify_graphs_trail(left: tuple[Triple, ...], right: tuple[Triple, ...]) -> bool:
if len(left) != len(right):
return False
used = [False] * len(right)
def step(index: int) -> bool:
if index >= len(left):
return True
current = left[index]
for candidate_index, candidate in enumerate(right):
if used[candidate_index]:
continue
if (
isinstance(current.p, Iri)
and isinstance(candidate.p, Iri)
and current.p.value != candidate.p.value
):
continue
mark = len(trail)
if unify_triple_trail(current, candidate):
used[candidate_index] = True
if step(index + 1):
return True
used[candidate_index] = False
undo_to(mark)
return False
return step(0)
def unify_term_trail(a: Term, b: Term) -> bool:
a = apply_subst_trail(a)
b = apply_subst_trail(b)
if isinstance(a, Var):
return bind_var(a, b)
if isinstance(b, Var):
return bind_var(b, a)
if isinstance(a, Iri) and a.value == RDF_NIL and isinstance(b, ListTerm) and not b.elems:
return True
if isinstance(b, Iri) and b.value == RDF_NIL and isinstance(a, ListTerm) and not a.elems:
return True
if a is b or a == b:
return True
if isinstance(a, Literal) and isinstance(b, Literal):
return self.literal_equivalent(a, b)
if isinstance(a, ListTerm) and isinstance(b, ListTerm):
left = a.elems
right = b.elems
size = len(left)
if size != len(right):
return False
if size == 0:
return True
if size == 1:
return unify_term_trail(left[0], right[0])
if size == 2:
return unify_term_trail(left[0], right[0]) and unify_term_trail(left[1], right[1])
if size == 3:
return (
unify_term_trail(left[0], right[0])
and unify_term_trail(left[1], right[1])
and unify_term_trail(left[2], right[2])
)
for left_item, right_item in zip(left, right):
if not unify_term_trail(left_item, right_item):
return False
return True
if isinstance(a, ListTerm):
recovered = self.rdf_collection_to_list(b)
if recovered is not None:
return unify_term_trail(a, ListTerm(recovered))
if isinstance(b, ListTerm):
recovered = self.rdf_collection_to_list(a)
if recovered is not None:
return unify_term_trail(ListTerm(recovered), b)
if isinstance(a, OpenListTerm) and isinstance(b, ListTerm):
if len(b.elems) < len(a.prefix):
return False
for x, y in zip(a.prefix, b.elems):
if not unify_term_trail(x, y):
return False
return bind_var(Var(a.tail_var), ListTerm(b.elems[len(a.prefix):]))
if isinstance(b, OpenListTerm) and isinstance(a, ListTerm):
return unify_term_trail(b, a)
if isinstance(a, OpenListTerm) and isinstance(b, OpenListTerm):
common = min(len(a.prefix), len(b.prefix))
for x, y in zip(a.prefix[:common], b.prefix[:common]):
if not unify_term_trail(x, y):
return False
if len(a.prefix) == len(b.prefix):
return bind_var(Var(a.tail_var), Var(b.tail_var))
if len(a.prefix) < len(b.prefix):
return bind_var(Var(a.tail_var), OpenListTerm(b.prefix[common:], b.tail_var))
return bind_var(Var(b.tail_var), OpenListTerm(a.prefix[common:], a.tail_var))
if isinstance(a, GraphTerm) and isinstance(b, GraphTerm):
return unify_graphs_trail(a.triples, b.triples)
return False
def unify_triple_trail(a: Triple, b: Triple) -> bool:
return (
unify_term_trail(a.p, b.p)
and unify_term_trail(a.s, b.s)
and unify_term_trail(a.o, b.o)
)
def apply_delta(delta: Subst) -> bool:
for name, value in list(delta.items()):
if not unify_term_trail(Var(name), value):
return False
return True
def answer_from_current() -> Subst:
answer: Subst = {}
for name in answer_vars:
value = apply_subst_trail(Var(name))
if not (isinstance(value, Var) and value.name == name):
answer[name] = value
return answer
visited_reset = object()
Frame = dict[str, Any]
stack: list[Frame] = [
{"kind": "node", "goals": list(goals), "depth": depth, "reorder": allow_reorder}
]
while stack:
frame = stack.pop()
kind = frame["kind"]
if kind == "undo":
undo_to(frame["subst_mark"])
undo_visited_to(frame["visited_mark"])
continue
if kind == "delta_iter":
deltas = frame["deltas"]
while frame["index"] < len(deltas):
delta = deltas[frame["index"]]
frame["index"] += 1
mark = len(trail)
if not apply_delta(delta):
undo_to(mark)
continue
if not frame["rest"]:
answer = answer_from_current()
if goal_memo_key is not None:
completed_answers.append(dict(answer))
yield answer
undo_to(mark)
continue
stack.append(frame)
stack.append({"kind": "undo", "subst_mark": mark, "visited_mark": len(visited_trail)})
stack.append({
"kind": "node",
"goals": frame["rest"],
"depth": frame["depth"] + 1,
"reorder": frame["reorder"],
})
break
continue
if kind in {"fact_iter", "rule_fact_iter", "memo_answer_iter"}:
items = frame["items"]
while frame["index"] < len(items):
item = items[frame["index"]]
frame["index"] += 1
mark = len(trail)
if not unify_triple_trail(frame["goal"], item):
undo_to(mark)
continue
if not frame["rest"]:
answer = answer_from_current()
if goal_memo_key is not None:
completed_answers.append(dict(answer))
yield answer
undo_to(mark)
continue
stack.append(frame)
stack.append({"kind": "undo", "subst_mark": mark, "visited_mark": len(visited_trail)})
stack.append({
"kind": "node",
"goals": frame["rest"],
"depth": frame["depth"] + 1,
"reorder": frame["reorder"],
})
break
continue
if kind == "rule_iter":
rules = frame["rules"]
while frame["index"] < len(rules):
rule = rules[frame["index"]]
frame["index"] += 1
if len(rule.premise) != 1:
continue
std = self.standardize_apart(rule)
mark = len(trail)
if not unify_triple_trail(frame["goal"], std.premise[0]):
undo_to(mark)
continue
body = list(std.conclusion)
if frame["goal_was_visited"] and any(
goal_key(apply_subst_triple_trail(premise)) in visited_counts
for premise in body
):
undo_to(mark)
continue
visited_mark = len(visited_trail)
push_visited(frame["goal_key"])
stack.append(frame)
stack.append({"kind": "undo", "subst_mark": mark, "visited_mark": visited_mark})
next_goals = body + frame["rest"]
if frame["rest"]:
next_goals = body + [(visited_reset, visited_mark)] + frame["rest"]
stack.append({
"kind": "node",
"goals": next_goals,
"depth": frame["depth"] + 1,
"reorder": False,
})
break
continue
goals_now = frame["goals"]
depth_now = frame["depth"]
reorder_now = frame["reorder"]
if depth_now > self.max_depth:
continue
if not goals_now:
answer = answer_from_current()
if goal_memo_key is not None:
completed_answers.append(dict(answer))
yield answer
continue
if isinstance(goals_now[0], tuple) and goals_now[0][0] is visited_reset:
undo_visited_to(goals_now[0][1])
stack.append({
"kind": "node",
"goals": goals_now[1:],
"depth": depth_now,
"reorder": reorder_now,
})
continue
selected = select_goal_index_trail(goals_now) if reorder_now else 0
first = apply_subst_triple_trail(goals_now[selected])
rest = goals_now[:selected] + goals_now[selected + 1:]
# A registered builtin owns its predicate and does not fall through
# to ordinary facts or backward rules.
if isinstance(first.p, Iri):
handler = get_builtin(first.p.value)
if handler is not None:
# The selected goal is already substitution-applied. Match
# Eyeling's hot path: builtins return only the new bindings
# introduced while evaluating this goal, not a full copy of
# the current proof state.
builtin_subst = (
subst_mut
if first.p.value in {
LOG_NS + "collectAllIn",
LOG_NS + "forAllIn",
LOG_NS + "includes",
LOG_NS + "notIncludes",
}
else {}
)
ctx = BuiltinContext(first, builtin_subst, self)
deltas = list(handler(ctx))
if deltas:
stack.append({
"kind": "delta_iter",
"deltas": deltas,
"index": 0,
"rest": rest,
"depth": depth_now,
"reorder": reorder_now,
})
continue
if first.p.value in {RDF_FIRST, RDF_REST}:
seen_lists: set[ListTerm] = set()
for fact in self.facts:
for term in (fact.s, fact.p, fact.o):
if isinstance(term, ListTerm) and term.elems:
seen_lists.add(term)
synthetic: list[Triple] = []
for collection in seen_lists:
obj = collection.elems[0] if first.p.value == RDF_FIRST else ListTerm(collection.elems[1:])
synthetic.append(Triple(collection, first.p, obj))
if synthetic:
stack.append({
"kind": "fact_iter",
"items": synthetic,
"index": 0,
"goal": first,
"rest": rest,
"depth": depth_now,
"reorder": reorder_now,
})
# Predicate-scoped memoization remains opt-in via log:memoize.
if isinstance(first.p, Iri) and first.p.value in self._memoized_predicates:
memo_key = self._predicate_memo_key(first)
if memo_key is not None:
table, memo_entry = self._predicate_memo_lookup(memo_key)
if not memo_entry["complete"] and not memo_entry["computing"]:
bottom_up_entry = self._try_bottom_up_numeric_memo(first)
if bottom_up_entry is not None:
memo_entry = bottom_up_entry
if memo_entry["complete"]:
stack.append({
"kind": "memo_answer_iter",
"items": memo_entry["answers"],
"index": 0,
"goal": first,
"rest": rest,
"depth": depth_now,
"reorder": reorder_now,
})
continue
if not memo_entry["computing"]:
memo_entry["computing"] = True
memo_successors: list[Subst] = []
try:
for nxt in self.solve(
[first],
{},
depth_now + 1,
reorder_now,
frozenset(visited_counts),
):
self._store_predicate_memo_answer(memo_entry, first, nxt)
memo_successors.append(nxt)
finally:
memo_entry["computing"] = False
if memo_entry["unsafe"]:
table.pop(memo_key, None)
else:
memo_entry["complete"] = True
if memo_successors:
stack.append({
"kind": "delta_iter",
"deltas": memo_successors,
"index": 0,
"rest": rest,
"depth": depth_now,
"reorder": reorder_now,
})
continue
# On re-entering an ancestor goal, reject only rules whose body
# immediately re-enters that ancestor chain. This is Eyeling's
# inexpensive guard for direct and mutual recursion.
first_key = goal_key(first)
goal_was_visited = first_key in visited_counts
# Eyeling only indexes/applies backward rules for a ground IRI
# predicate. Variable-predicate goals range over facts.
candidate_rules: list[Rule]
if isinstance(first.p, Iri):
candidate_rules = [
*self._backward_rules_by_pred.get(first.p.value, ()),
*self._wild_backward_rules,
]
else:
candidate_rules = []
# Push in reverse processing order so facts are explored first, as
# in the previous Python solver.
if candidate_rules:
stack.append({
"kind": "rule_iter",
"rules": candidate_rules,
"index": 0,
"goal": first,
"goal_key": first_key,
"goal_was_visited": goal_was_visited,
"rest": rest,
"depth": depth_now,
})
rule_facts = list(self._candidate_rule_facts(first))
if rule_facts:
stack.append({
"kind": "rule_fact_iter",
"items": rule_facts,
"index": 0,
"goal": first,
"rest": rest,
"depth": depth_now,
"reorder": reorder_now,
})
facts = list(self._candidate_facts(first))
if facts:
stack.append({
"kind": "fact_iter",
"items": facts,
"index": 0,
"goal": first,
"rest": rest,
"depth": depth_now,
"reorder": reorder_now,
})
if goal_memo is not None and goal_memo_key is not None:
goal_memo[goal_memo_key] = completed_answers