"""Deterministic evidence verification; model scores never override graph truth.""" from __future__ import annotations from dataclasses import replace from graph.constraints import path_satisfies_constraints from graph.graph import GraphStore from graph.types import ConstraintState, GraphPath, GraphQuery, ReasoningStatus def verify_path( path: GraphPath, graph: GraphStore, query: GraphQuery | ConstraintState ) -> GraphPath: constraints = ( query if isinstance(query, ConstraintState) else ConstraintState.from_query(query) ) valid = bool(path.steps) for index, step in enumerate(path.steps): valid = ( valid and step.source_id in graph.nodes and step.target_id in graph.nodes ) valid = valid and graph.has_edge( step.source_id, step.relation_id, step.target_id ) if step.source_type is not None: valid = valid and graph.nodes.get(step.source_id) == step.source_type if step.target_type is not None: valid = valid and graph.nodes.get(step.target_id) == step.target_type if index and path.steps[index - 1].target_id != step.source_id: valid = False valid = valid and path_satisfies_constraints(path, constraints, graph) return replace(path, verified=valid) def classify_evidence( *, candidate_paths: tuple[GraphPath, ...], graph: GraphStore, query: GraphQuery, ) -> ReasoningStatus: verified = tuple(verify_path(path, graph, query) for path in candidate_paths) valid = tuple(path for path in verified if path.verified) if not valid: return ReasoningStatus.INSUFFICIENT_EVIDENCE return ReasoningStatus.SUPPORTED def detect_conflicting_paths( paths: tuple[GraphPath, ...], graph: GraphStore | None = None ) -> tuple[tuple[GraphPath, ...], ...]: """Return groups of paths that assert different values for the same claim. Three conditions must hold for a contradiction (PLAN section 18): 1. the paths assert a *single* fact, so only one-step claims are grouped -- a multi-hop path is an explanation chain, and two chains reaching different endpoints are competing answers, not a conflict; 2. they share a subject and relation; 3. that relation is **functional**. ``owns`` and ``uses`` are set-valued, so two values are two facts; ``works_at`` is single-valued, so two values cannot both hold. Cardinality is read from the graph, which is the source of truth. When the graph declares nothing, no contradiction is asserted: silence must not be read as a conflict. """ groups: dict[tuple[str, str], list[GraphPath]] = {} for path in paths: if len(path.steps) != 1: continue first = path.steps[0] if not _is_functional(graph, first.relation_id): continue groups.setdefault((first.source_id, first.relation_id), []).append(path) return tuple( tuple(group) for group in groups.values() if len({path.endpoint_id for path in group}) > 1 ) def _is_functional(graph: GraphStore | None, relation_id: str) -> bool: if graph is None: return False attributes = graph.relation_attributes.get(relation_id, {}) return str(attributes.get("functional", "false")).lower() == "true" def find_contradictions( graph: GraphStore, ) -> tuple[tuple[str, str, tuple[str, ...]], ...]: """Enumerate functional relations in ``graph`` that assert multiple values. Used by the evaluator and the API to report conflicts that exist in the graph itself, independently of which paths a search happened to return. """ claims: dict[tuple[str, str], set[str]] = {} for edge in graph.edges: if not _is_functional(graph, edge.relation_id): continue claims.setdefault((edge.source_id, edge.relation_id), set()).add(edge.target_id) return tuple( (source, relation, tuple(sorted(targets))) for (source, relation), targets in sorted(claims.items()) if len(targets) > 1 )