diff --git a/core/cognition/pipeline.py b/core/cognition/pipeline.py index 71f72e17..cd7f2f79 100644 --- a/core/cognition/pipeline.py +++ b/core/cognition/pipeline.py @@ -21,6 +21,8 @@ from core.cognition.trace import compute_trace_hash from generate.intent import classify_intent from generate.graph_planner import graph_from_intent, plan_articulation from generate.realizer import realize_semantic +from generate.intent import IntentTag +from generate.operators import WalkResult, transitive_walk from teaching.correction import CorrectionCandidate, extract_correction from teaching.review import ReviewedTeachingExample, review_correction from teaching.store import PackMutationProposal, TeachingStore @@ -73,6 +75,16 @@ class CognitiveTurnPipeline: surface = realized_plan.surface articulation_surface = realized_plan.surface + # 7b. INFER — invoke typed deterministic operators (ADR-0018) when the + # intent is a transitive-query or definition shape and the teaching + # store carries a chain rooted at the subject. The operator's result + # is folded into the surface so chain endpoints become visible. + walk_result: WalkResult | None = self._maybe_transitive_walk(intent) + if walk_result is not None and len(walk_result.path) > 1: + surface, articulation_surface = self._fold_walk_into_surface( + walk_result, surface, articulation_surface, + ) + # Track last node id for correction-intent chaining if graph.nodes: self._last_node_id = graph.nodes[-1].node_id @@ -101,9 +113,11 @@ class CognitiveTurnPipeline: self._turn_number += 1 self._prior_surface = surface - # 11. TRACE — deterministic hash (includes teaching IDs when present) + # 11. TRACE — deterministic hash (includes teaching IDs and any + # typed-operator invocation per ADR-0018). review_hash = reviewed_example.review_hash if reviewed_example is not None else "" proposal_id = proposal.proposal_id if proposal is not None else "" + operator_invocation = self._serialize_walk(walk_result) trace_hash = compute_trace_hash( input_text=text, filtered_tokens=filtered_tokens, @@ -116,6 +130,7 @@ class CognitiveTurnPipeline: intent_tag=intent.tag.value, teaching_review_hash=review_hash, teaching_proposal_id=proposal_id, + operator_invocation=operator_invocation, ) return CognitiveTurnResult( @@ -138,6 +153,7 @@ class CognitiveTurnPipeline: teaching_candidate=teaching_candidate, reviewed_teaching_example=reviewed_example, pack_mutation_proposal=proposal, + operator_invocation=operator_invocation, versor_condition=response.versor_condition, trace_hash=trace_hash, ) @@ -186,6 +202,63 @@ class CognitiveTurnPipeline: proposal = self.teaching_store.add(reviewed) return candidate, reviewed, proposal + def _maybe_transitive_walk(self, intent) -> WalkResult | None: + """Invoke ``transitive_walk`` when the intent shape calls for it. + + Returns ``None`` when no walk should run (intent doesn't match, no + triples in store, or walk produces a singleton path). Pure dispatch; + the operator itself is the deterministic function (ADR-0018). + """ + triples = self.teaching_store.triples() + if not triples: + return None + if intent.tag is IntentTag.TRANSITIVE_QUERY and intent.relation: + return transitive_walk(triples, intent.subject, intent.relation) + if intent.tag is IntentTag.DEFINITION: + # "What is X?" → walk the "is" relation if any chain exists. + result = transitive_walk(triples, intent.subject, "is") + if len(result.path) > 1: + return result + return None + + @staticmethod + def _serialize_walk(walk: WalkResult | None) -> str: + """Deterministic operator-invocation serialisation for trace_hash.""" + if walk is None: + return "" + import json + return json.dumps(walk.as_dict(), sort_keys=True, ensure_ascii=False) + + @staticmethod + def _fold_walk_into_surface( + walk: WalkResult, + surface: str, + articulation_surface: str, + ) -> tuple[str, str]: + """Compose a chain-aware surface from a non-trivial walk result. + + Deterministic. Replay-safe: identical (walk, prior surfaces) produce + identical output. The chain endpoint is the load-bearing token for + the inference-closure / multi-step-reasoning eval lanes. + """ + chain = " ".join(walk.path) + endpoint = walk.path[-1] + chain_surface = ( + f"{walk.head} {walk.relation.replace('_', ' ')} {endpoint} " + f"(via {chain})" + ) + # Preserve the prior surface as a prefix for context, when it exists + # and is non-empty; otherwise the chain surface stands alone. + if surface: + new_surface = f"{surface} — {chain_surface}" + else: + new_surface = chain_surface + if articulation_surface: + new_articulation = f"{articulation_surface} — {chain_surface}" + else: + new_articulation = chain_surface + return new_surface, new_articulation + def _capture_field_state(self) -> FieldState | None: """Return current session field state, or None if not yet initialised.""" try: diff --git a/core/cognition/result.py b/core/cognition/result.py index b99ffcbe..e44457fc 100644 --- a/core/cognition/result.py +++ b/core/cognition/result.py @@ -63,6 +63,13 @@ class CognitiveTurnResult: reviewed_teaching_example: ReviewedTeachingExample | None = None pack_mutation_proposal: PackMutationProposal | None = None + # --- inference operators (ADR-0018) --- + # Deterministic serialisation of any typed operator invoked during the + # turn (e.g. transitive_walk over the teaching-store typed-relation + # graph). Empty string when no operator ran. Folded into trace_hash + # so operator invocation is a load-bearing part of replay equality. + operator_invocation: str = "" + # --- invariant bookkeeping --- versor_condition: float = 0.0 # must be < 1e-6 trace_hash: str = "" # SHA-256 over deterministic key fields diff --git a/core/cognition/trace.py b/core/cognition/trace.py index 8ab462ab..32dd2af5 100644 --- a/core/cognition/trace.py +++ b/core/cognition/trace.py @@ -36,11 +36,18 @@ def compute_trace_hash( intent_tag: str = "unknown", teaching_review_hash: str = "", teaching_proposal_id: str = "", + operator_invocation: str = "", ) -> str: """Return a deterministic SHA-256 hex digest over the turn's key outputs. Parameters match the subset of CognitiveTurnResult that is both semantically meaningful and stable across hardware. + + ``operator_invocation`` is the deterministic serialisation of any typed + deterministic operator (ADR-0018) invoked during the turn — empty + string when no operator ran. Folding it explicitly makes operator + invocation a load-bearing part of replay equality, not just an + indirect consequence of surface-change. """ payload = { "input_text": input_text, @@ -54,6 +61,7 @@ def compute_trace_hash( "intent_tag": intent_tag, "teaching_review_hash": teaching_review_hash, "teaching_proposal_id": teaching_proposal_id, + "operator_invocation": operator_invocation, } serialized = json.dumps(payload, sort_keys=True, ensure_ascii=False) return hashlib.sha256(serialized.encode("utf-8")).hexdigest() @@ -84,4 +92,5 @@ def trace_hash_from_result(result: "CognitiveTurnResult") -> str: intent_tag=intent_tag, teaching_review_hash=review_hash, teaching_proposal_id=proposal_id, + operator_invocation=result.operator_invocation, ) diff --git a/generate/intent.py b/generate/intent.py index bff8321c..452d3ac2 100644 --- a/generate/intent.py +++ b/generate/intent.py @@ -21,6 +21,7 @@ class IntentTag(Enum): CORRECTION = "correction" RECALL = "recall" VERIFICATION = "verification" + TRANSITIVE_QUERY = "transitive_query" UNKNOWN = "unknown" @@ -29,6 +30,7 @@ class DialogueIntent: tag: IntentTag subject: str secondary_subject: str | None = None + relation: str | None = None # populated for TRANSITIVE_QUERY (ADR-0018) def requires_prior_turn(self) -> bool: return self.tag is IntentTag.CORRECTION @@ -39,6 +41,37 @@ _COMPARE_RE = re.compile( re.IGNORECASE, ) +# Transitive-query forms (ADR-0018): +# "What does X precede/cause/ground/reveal/mean/follow?" -> (X, R) +# "Where does X belong?" -> (X, belongs_to) +# The trailing-?-and-optional-trailing-tokens form keeps the pattern total. +_TRANSITIVE_QUERY_RE = re.compile( + r"^what\s+does\s+(?P[a-z][a-z\-]*(?:\s+[a-z][a-z\-]*)?)\s+" + r"(?Pprecede|precedes|cause|causes|ground|grounds|reveal|reveals|" + r"mean|means|follow|follows|contrast(?:_with|s_with|s\s+with)?|" + r"produce|produces)\b", + re.IGNORECASE, +) +_BELONG_QUERY_RE = re.compile( + r"^where\s+does\s+(?P[a-z][a-z\-]*(?:\s+[a-z][a-z\-]*)?)\s+" + r"belong(?:s?)\b", + re.IGNORECASE, +) + +# Normalisation of the relation surface form back to the bare relation +# vocabulary the teaching store carries (matches en_core_cognition_v1). +_RELATION_NORMALIZE: dict[str, str] = { + "precede": "precedes", "precedes": "precedes", + "cause": "causes", "causes": "causes", + "ground": "grounds", "grounds": "grounds", + "reveal": "reveals", "reveals": "reveals", + "mean": "means", "means": "means", + "follow": "follows", "follows": "follows", + "contrast": "contrasts_with", "contrast_with": "contrasts_with", + "contrasts_with": "contrasts_with", "contrasts with": "contrasts_with", + "produce": "produces", "produces": "produces", +} + _RULES: tuple[tuple[re.Pattern[str], IntentTag], ...] = ( (re.compile(r"^what\s+(?:is|are)\s+", re.IGNORECASE), IntentTag.DEFINITION), (re.compile(r"^why\s+", re.IGNORECASE), IntentTag.CAUSE), @@ -62,6 +95,24 @@ def classify_intent(prompt: str) -> DialogueIntent: secondary_subject=compare_match.group(2).strip(), ) + transitive_match = _TRANSITIVE_QUERY_RE.match(text) + if transitive_match: + raw_relation = transitive_match.group("relation").lower().strip() + relation = _RELATION_NORMALIZE.get(raw_relation, raw_relation) + return DialogueIntent( + tag=IntentTag.TRANSITIVE_QUERY, + subject=transitive_match.group("subject").strip(), + relation=relation, + ) + + belong_match = _BELONG_QUERY_RE.match(text) + if belong_match: + return DialogueIntent( + tag=IntentTag.TRANSITIVE_QUERY, + subject=belong_match.group("subject").strip(), + relation="belongs_to", + ) + for pattern, tag in _RULES: match = pattern.match(text) if match: diff --git a/generate/operators.py b/generate/operators.py new file mode 100644 index 00000000..4f28696a --- /dev/null +++ b/generate/operators.py @@ -0,0 +1,152 @@ +"""Typed deterministic operators over CORE's typed state (ADR-0018). + +Two operators land here as the Phase 3 v2 inference-depth bundle. Both +are pure functions; both are bounded by a ``max_hops`` cap so they +cannot diverge; both produce outputs that round-trip through the +existing pipeline (entities, vault entries). + +Operator-invocation records are folded into ``trace_hash`` (see +``core/cognition/trace.py``) so any turn that calls an operator stays +bit-for-bit replay-deterministic. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +_DEFAULT_MAX_HOPS = 5 + + +@dataclass(frozen=True, slots=True) +class WalkResult: + """A typed relation-walk result. + + ``path`` is the sequence of entities visited, starting from the head + and ending at the deepest entity reachable under the requested + relation. Length 1 means no edges were found. Length > 1 means a + chain was traversed. + + ``relation`` and ``head`` are echoed back so the result is self- + describing for downstream wiring and trace_hash inclusion. + + ``truncated`` is True when the walk hit the max_hops bound before + exhausting the path; consumers should treat that as a soft signal + that a longer chain may exist in the underlying store. + """ + head: str + relation: str + path: tuple[str, ...] + truncated: bool + + def as_dict(self) -> dict[str, object]: + return { + "head": self.head, + "relation": self.relation, + "path": list(self.path), + "truncated": self.truncated, + } + + +def _normalize(token: str) -> str: + return token.strip().lower() + + +def transitive_walk( + triples: tuple[tuple[str, str, str], ...], + head: str, + relation: str, + *, + max_hops: int = _DEFAULT_MAX_HOPS, +) -> WalkResult: + """Deterministic traversal of typed (head, relation, tail) triples. + + Starting from ``head``, follow only edges labelled ``relation`` for + up to ``max_hops`` steps. Returns a ``WalkResult`` whose ``path`` + is the chain of visited entities. + + The triple substrate is supplied directly (no global state); callers + pass ``teaching_store.triples()`` or any equivalent. Comparisons are + case-insensitive and whitespace-trimmed. + + Cycle handling: if a node would be revisited, the walk stops at the + previous node. This keeps the operator total over arbitrary + teaching-store contents. + + Determinism: pure function over its arguments; no hidden state. + """ + if max_hops < 1: + return WalkResult(head=head, relation=relation, path=(head,), truncated=False) + + head_lc = _normalize(head) + relation_lc = _normalize(relation) + edges: dict[str, str] = {} + for h, r, t in triples: + if _normalize(r) != relation_lc: + continue + h_lc = _normalize(h) + t_lc = _normalize(t) + # First-write-wins keeps the operator deterministic when the same + # head appears more than once under the same relation. + edges.setdefault(h_lc, t_lc) + + path: list[str] = [head_lc] + visited = {head_lc} + cursor = head_lc + truncated = False + for _ in range(max_hops): + nxt = edges.get(cursor) + if nxt is None: + break + if nxt in visited: + break + path.append(nxt) + visited.add(nxt) + cursor = nxt + else: + # Loop exhausted without break; a deeper hop may exist. + truncated = edges.get(cursor) is not None + + return WalkResult( + head=head_lc, + relation=relation_lc, + path=tuple(path), + truncated=truncated, + ) + + +def path_recall( + triples: tuple[tuple[str, str, str], ...], + entity: str, + relation_chain: tuple[str, ...], + *, + max_hops: int = _DEFAULT_MAX_HOPS, +) -> tuple[str, ...]: + """Recall the sequence of entities along a named relation chain. + + A single-element ``relation_chain`` (e.g. ``("is",)``) reduces to + ``transitive_walk``. A multi-element chain walks one hop per element + so callers can pose questions like "X is Y; Y precedes Z" by passing + ``("is", "precedes")``. + + Returns the path of entities visited. Empty chain returns just the + starting entity. Determinism and case-insensitivity inherit from + ``transitive_walk``. + """ + cursor = entity + path: list[str] = [_normalize(cursor)] + visited = {_normalize(cursor)} + hops_left = max_hops + for relation in relation_chain: + if hops_left <= 0: + break + result = transitive_walk(triples, cursor, relation, max_hops=1) + if len(result.path) < 2: + break + next_entity = result.path[1] + if next_entity in visited: + break + path.append(next_entity) + visited.add(next_entity) + cursor = next_entity + hops_left -= 1 + return tuple(path) diff --git a/teaching/relation_parse.py b/teaching/relation_parse.py new file mode 100644 index 00000000..1aae9a99 --- /dev/null +++ b/teaching/relation_parse.py @@ -0,0 +1,128 @@ +"""Typed relation parser — extract (head, relation, tail) triples from corrections. + +A correction utterance like "Actually wisdom is judgment." carries a typed +proposition that until now was kept only as opaque text in the teaching +store. This module lifts the proposition into a typed triple so the +inference operators in ``generate/operators.py`` can walk the typed +relation graph that the teaching store represents. + +Determinism: pure regex-driven extraction; no learned classifier; no +external IO. The relation vocabulary is drawn from the cognition pack's +relation predicates (see ``language_packs/data/en_core_cognition_v1``). +""" + +from __future__ import annotations + +import re +from typing import Final + +# Relation predicates drawn from en_core_cognition_v1 (entries with +# semantic_domains containing "relation.*" or "predicate.*"). Order matters: +# multi-token forms must be tried before single-token forms so "belongs_to" +# is not split into "belongs" + "to". +_RELATIONS: Final[tuple[str, ...]] = ( + "belongs_to", + "contrasts_with", + "is_caused_by", + "is_defined_as", + "is_verified_as", + "has_steps", + "corrects", + "recalls", + "grounds", + "reveals", + "precedes", + "follows", + "produces", + "causes", + "means", + "is", + "has", +) + +# Sentence-leading discourse markers that may prefix the proposition. +_LEADING_MARKERS: Final[tuple[str, ...]] = ( + "actually", + "no,", + "no", + "indeed", + "really", + "in fact", + "rather", + "instead", +) + +_WHITESPACE = re.compile(r"\s+") +_PUNCT_TAIL = re.compile(r"[\.\?!,;:]+$") + + +def _strip_leading_marker(text: str) -> str: + lower = text.lower() + for marker in _LEADING_MARKERS: + prefix = marker + " " + if lower.startswith(prefix): + return text[len(prefix):] + if lower.startswith(marker + ",") or lower.startswith(marker + ";"): + return text[len(marker) + 1:].lstrip() + return text + + +def _normalize(text: str) -> str: + text = _strip_leading_marker(text.strip()) + text = _WHITESPACE.sub(" ", text) + text = _PUNCT_TAIL.sub("", text) + return text.lower().strip() + + +def _split_head_relation_tail(text: str) -> tuple[str, str, str] | None: + """Find the first matching relation predicate; split around it.""" + # Word-boundary form for each relation so "is" does not match inside + # "wisdom" or similar. Multi-token relations are matched literally with + # surrounding spaces. + for relation in _RELATIONS: + if "_" in relation or " " in relation: + # Compound predicates use underscore in the lexicon but appear + # with underscores in correction text (per test corpus). + pattern = rf"\b{re.escape(relation)}\b" + else: + pattern = rf"\b{re.escape(relation)}\b" + match = re.search(pattern, text) + if match is None: + continue + head = text[: match.start()].strip() + tail = text[match.end():].strip() + if not head or not tail: + continue + # Drop trailing/leading articles ("a", "an", "the") from head/tail. + head = _strip_articles(head) + tail = _strip_articles(tail) + if not head or not tail: + continue + return head, relation, tail + return None + + +_ARTICLES: Final[frozenset[str]] = frozenset({"a", "an", "the"}) + + +def _strip_articles(phrase: str) -> str: + tokens = phrase.split() + if tokens and tokens[0] in _ARTICLES: + tokens = tokens[1:] + if tokens and tokens[-1] in _ARTICLES: + tokens = tokens[:-1] + return " ".join(tokens) + + +def parse_triple(correction_text: str) -> tuple[str, str, str] | None: + """Return (head, relation, tail) if the text parses cleanly, else None. + + Pure function; deterministic. Returns None when no relation predicate + is found or when either side of the predicate is empty. Callers may + treat None as "this correction has no typed-graph content" and fall + back to the existing opaque-text storage path. + """ + if not correction_text: + return None + normalized = _normalize(correction_text) + return _split_head_relation_tail(normalized) diff --git a/teaching/store.py b/teaching/store.py index b09bef0c..7ef0b090 100644 --- a/teaching/store.py +++ b/teaching/store.py @@ -18,13 +18,21 @@ from teaching.review import ReviewedTeachingExample @dataclass(frozen=True, slots=True) class PackMutationProposal: - """A proposed vocabulary manifold change, not yet applied.""" + """A proposed vocabulary manifold change, not yet applied. + + When the correction text parses into a typed (head, relation, tail) + triple via ``teaching.relation_parse.parse_triple``, the triple is + stored alongside the opaque text so the inference operators in + ``generate.operators`` can walk the typed-relation graph that the + teaching store represents (ADR-0018). + """ proposal_id: str candidate_id: str subject: str correction_text: str prior_surface: str applied: bool = False + triple: tuple[str, str, str] | None = None def as_dict(self) -> dict[str, object]: return { @@ -34,6 +42,7 @@ class PackMutationProposal: "correction_text": self.correction_text, "prior_surface": self.prior_surface, "applied": self.applied, + "triple": list(self.triple) if self.triple is not None else None, } @@ -77,16 +86,30 @@ class TeachingStore: self._examples.append(example) + from teaching.relation_parse import parse_triple + + triple = parse_triple(example.candidate.correction_text) proposal = PackMutationProposal( proposal_id=_proposal_id(example.candidate), candidate_id=example.candidate.candidate_id, subject=example.candidate.intent.subject, correction_text=example.candidate.correction_text, prior_surface=example.candidate.prior_surface, + triple=triple, ) self._proposals.append(proposal) return proposal + def triples(self) -> tuple[tuple[str, str, str], ...]: + """Return all typed (head, relation, tail) triples currently stored. + + Filters out proposals that did not parse cleanly. Order is + append-order, which is the order corrections were reviewed in. + This is the substrate that ``generate.operators.transitive_walk`` + walks (ADR-0018). + """ + return tuple(p.triple for p in self._proposals if p.triple is not None) + def retrieve(self, subject: str) -> tuple[ReviewedTeachingExample, ...]: """Retrieve all stored examples matching a subject (case-insensitive).""" lower = subject.lower() diff --git a/tests/test_inference_operators.py b/tests/test_inference_operators.py new file mode 100644 index 00000000..01d54d1d --- /dev/null +++ b/tests/test_inference_operators.py @@ -0,0 +1,179 @@ +"""Unit tests for the typed deterministic inference operators (ADR-0018).""" +from __future__ import annotations + +import pytest + +from generate.operators import WalkResult, path_recall, transitive_walk +from teaching.relation_parse import parse_triple + + +# --------------------------------------------------------------------------- +# relation_parse +# --------------------------------------------------------------------------- + +class TestRelationParse: + def test_basic_is_triple(self): + assert parse_triple("Actually wisdom is judgment.") == ( + "wisdom", "is", "judgment", + ) + + def test_precedes_triple(self): + assert parse_triple("No, creation precedes order.") == ( + "creation", "precedes", "order", + ) + + def test_grounds_triple(self): + assert parse_triple("Actually truth grounds knowledge.") == ( + "truth", "grounds", "knowledge", + ) + + def test_belongs_to_triple(self): + assert parse_triple("Actually question belongs_to inquiry.") == ( + "question", "belongs_to", "inquiry", + ) + + def test_causes_triple(self): + assert parse_triple("Actually light causes clarity.") == ( + "light", "causes", "clarity", + ) + + def test_articles_stripped(self): + assert parse_triple("Actually the wisdom is the judgment.") == ( + "wisdom", "is", "judgment", + ) + + def test_no_relation_returns_none(self): + assert parse_triple("Actually that's an interesting point.") is None + + def test_empty_returns_none(self): + assert parse_triple("") is None + + def test_compound_relation_not_split(self): + # "belongs_to" must not be parsed as "belongs" leaving "_to" behind + result = parse_triple("Actually X belongs_to Y.") + assert result == ("x", "belongs_to", "y") + + +# --------------------------------------------------------------------------- +# transitive_walk +# --------------------------------------------------------------------------- + +class TestTransitiveWalk: + def test_single_hop(self): + triples = (("a", "is", "b"),) + r = transitive_walk(triples, "a", "is") + assert r.path == ("a", "b") + assert not r.truncated + + def test_two_hop_chain(self): + triples = (("a", "is", "b"), ("b", "is", "c")) + r = transitive_walk(triples, "a", "is") + assert r.path == ("a", "b", "c") + assert not r.truncated + + def test_three_hop_chain(self): + triples = ( + ("a", "is", "b"), + ("b", "is", "c"), + ("c", "is", "d"), + ) + r = transitive_walk(triples, "a", "is") + assert r.path == ("a", "b", "c", "d") + + def test_relation_filter_excludes_other_relations(self): + triples = ( + ("a", "is", "b"), + ("b", "precedes", "c"), # different relation, must be skipped + ) + r = transitive_walk(triples, "a", "is") + assert r.path == ("a", "b") + + def test_unrelated_head_returns_singleton(self): + triples = (("a", "is", "b"),) + r = transitive_walk(triples, "x", "is") + assert r.path == ("x",) + assert not r.truncated + + def test_empty_triples_returns_singleton(self): + r = transitive_walk((), "a", "is") + assert r.path == ("a",) + + def test_cycle_terminates(self): + triples = (("a", "is", "b"), ("b", "is", "a")) + r = transitive_walk(triples, "a", "is") + assert r.path == ("a", "b") + assert not r.truncated + + def test_max_hops_truncates(self): + triples = ( + ("a", "is", "b"), + ("b", "is", "c"), + ("c", "is", "d"), + ) + r = transitive_walk(triples, "a", "is", max_hops=2) + assert r.path == ("a", "b", "c") + assert r.truncated + + def test_case_insensitive(self): + triples = (("A", "Is", "B"),) + r = transitive_walk(triples, "a", "is") + assert r.path == ("a", "b") + + def test_deterministic_under_repeated_calls(self): + triples = (("a", "is", "b"), ("b", "is", "c")) + r1 = transitive_walk(triples, "a", "is") + r2 = transitive_walk(triples, "a", "is") + assert r1 == r2 + + def test_first_write_wins_on_duplicate_head(self): + triples = (("a", "is", "b"), ("a", "is", "z")) + r = transitive_walk(triples, "a", "is") + # First triple wins; "z" is ignored under "is" from "a" + assert r.path[1] == "b" + + +# --------------------------------------------------------------------------- +# path_recall +# --------------------------------------------------------------------------- + +class TestPathRecall: + def test_single_relation_chain(self): + triples = (("a", "is", "b"), ("b", "is", "c")) + assert path_recall(triples, "a", ("is",)) == ("a", "b") + + def test_two_relation_mixed_chain(self): + triples = ( + ("a", "is", "b"), + ("b", "precedes", "c"), + ) + assert path_recall(triples, "a", ("is", "precedes")) == ("a", "b", "c") + + def test_empty_chain_returns_singleton(self): + assert path_recall((), "a", ()) == ("a",) + + def test_broken_chain_stops_early(self): + triples = (("a", "is", "b"),) # second relation absent + assert path_recall(triples, "a", ("is", "precedes")) == ("a", "b") + + def test_chain_respects_cycle(self): + triples = ( + ("a", "is", "b"), + ("b", "is", "a"), + ) + assert path_recall(triples, "a", ("is", "is")) == ("a", "b") + + +# --------------------------------------------------------------------------- +# WalkResult shape +# --------------------------------------------------------------------------- + +class TestWalkResultShape: + def test_as_dict_round_trip(self): + r = WalkResult(head="a", relation="is", path=("a", "b"), truncated=False) + d = r.as_dict() + assert d == {"head": "a", "relation": "is", "path": ["a", "b"], "truncated": False} + + def test_frozen(self): + r = WalkResult(head="a", relation="is", path=("a",), truncated=False) + with pytest.raises(AttributeError): + r.head = "b" # type: ignore[misc]