diff --git a/mempalace/fact_checker.py b/mempalace/fact_checker.py new file mode 100644 index 0000000000..b6b4df1fd3 --- /dev/null +++ b/mempalace/fact_checker.py @@ -0,0 +1,402 @@ +""" +fact_checker.py — Rule-based contradiction detection for MemPalace. +==================================================================== + +Checks assertions in natural language against the knowledge graph to +detect factual conflicts. No API calls, no external dependencies — +pure rule-based pattern matching + KG lookup. + +Severity levels: + RED — Direct factual contradiction (wrong person, wrong relationship) + YELLOW — Numeric/temporal mismatch (wrong tenure, wrong date) + GREEN — No conflict found (consistent or no data to compare) + +Usage: + from mempalace.fact_checker import check_assertion + from mempalace.knowledge_graph import KnowledgeGraph + + kg = KnowledgeGraph() + result = check_assertion("Soren finished the auth migration", kg) + # => CheckResult(severity="GREEN", ...) or RED/YELLOW with explanation + +Works with the MCP server via the mempalace_check_facts tool. +""" + +import re +from dataclasses import dataclass, field +from datetime import date +from typing import List, Optional + + +# ── Result types ───────────────────────────────────────────────────── + + +@dataclass +class Conflict: + """A single conflict between an assertion and the knowledge graph.""" + + severity: str # "RED" or "YELLOW" + entity: str + field: str # e.g. "attribution", "tenure", "role", "relationship" + message: str + kg_fact: Optional[dict] = None + + +@dataclass +class CheckResult: + """Result of checking an assertion against the knowledge graph.""" + + severity: str # "RED", "YELLOW", or "GREEN" + text: str # the original assertion + conflicts: List[Conflict] = field(default_factory=list) + entities_checked: List[str] = field(default_factory=list) + + def to_dict(self) -> dict: + """Serialize for MCP tool response.""" + result = { + "severity": self.severity, + "text": self.text, + "entities_checked": self.entities_checked, + "conflicts": [], + } + for c in self.conflicts: + entry = { + "severity": c.severity, + "entity": c.entity, + "field": c.field, + "message": c.message, + } + if c.kg_fact: + entry["kg_fact"] = c.kg_fact + result["conflicts"].append(entry) + return result + + +# ── Assertion patterns ─────────────────────────────────────────────── +# Each pattern extracts (subject, claim_type, object) from natural language. + +# Attribution: "X finished/completed/did/built/wrote Y" +_ATTRIBUTION_PATTERN = re.compile( + r"\b([A-Z][a-z]+(?:\s[A-Z][a-z]+)?)\s+" + r"(?:finished|completed|did|built|wrote|shipped|deployed|created|designed|implemented|fixed)\s+" + r"(?:the\s+)?(.+?)(?:\.|$)", + re.IGNORECASE, +) + +# Tenure: "X has been here N years" or "X joined N years ago" +_TENURE_PATTERN = re.compile( + r"\b([A-Z][a-z]+(?:\s[A-Z][a-z]+)?)\s+" + r"(?:has been (?:here|at \w+)|(?:has )?worked (?:here|at \w+))\s+" + r"(?:for\s+)?(\d+)\s+years?", + re.IGNORECASE, +) + +# Role: "X is a/the Y" (where Y is a role-like word) +_ROLE_PATTERN = re.compile( + r"\b([A-Z][a-z]+(?:\s[A-Z][a-z]+)?)\s+is\s+(?:a|the|an)\s+" + r"([\w\s]+?)(?:\s+(?:at|of|for|in)\b.*)?(?:\.|,|$)", + re.IGNORECASE, +) + +# Relationship: "X is Y's Z" (e.g. "Max is Alice's son") +_RELATIONSHIP_PATTERN = re.compile( + r"\b([A-Z][a-z]+(?:\s[A-Z][a-z]+)?)\s+is\s+" + r"([A-Z][a-z]+(?:\s[A-Z][a-z]+)?)'s\s+" + r"(daughter|son|child|mother|father|parent|wife|husband|partner|brother|sister|sibling|pet|dog|cat)", + re.IGNORECASE, +) + +# Relationship predicates that map to each other +_RELATIONSHIP_PAIRS = { + "daughter": ("child_of", "is_child_of", "parent_of"), + "son": ("child_of", "is_child_of", "parent_of"), + "child": ("child_of", "is_child_of", "parent_of"), + "mother": ("parent_of", "is_child_of", "child_of"), + "father": ("parent_of", "is_child_of", "child_of"), + "parent": ("parent_of", "is_child_of", "child_of"), + "wife": ("married_to", "is_partner_of"), + "husband": ("married_to", "is_partner_of"), + "partner": ("married_to", "is_partner_of"), + "brother": ("sibling_of", "is_sibling_of"), + "sister": ("sibling_of", "is_sibling_of"), + "sibling": ("sibling_of", "is_sibling_of"), + "pet": ("is_pet_of",), + "dog": ("is_pet_of",), + "cat": ("is_pet_of",), +} + +# Attribution predicates in the KG +_ATTRIBUTION_PREDICATES = { + "assigned_to", + "works_on", + "responsible_for", + "owns", + "leads", + "built", + "created", + "finished", + "completed", +} + + +# Negation words that invalidate a claim (suggested by @web3guru888) +_NEGATION_WORDS = ("not ", "no longer ", "never ", "didn't ", "doesn't ", "isn't ", "wasn't ") + + +def _is_negated(text: str, match_start: int, match_end: int = None) -> bool: + """Check if a regex match is negated. + + Checks both the prefix (30 chars before the match) and the match span + itself, since negation words often appear between subject and verb + (e.g. "Soren did NOT finish..."). + """ + prefix = text[:match_start].lower()[-30:] + if any(neg in prefix for neg in _NEGATION_WORDS): + return True + if match_end is not None: + span = text[match_start:match_end].lower() + return any(neg in span for neg in _NEGATION_WORDS) + return False + + +# ── Core checker ───────────────────────────────────────────────────── + + +def _check_attribution(text: str, kg) -> List[Conflict]: + """Check if an attribution claim conflicts with the KG.""" + conflicts = [] + match = _ATTRIBUTION_PATTERN.search(text) + if not match: + return conflicts + + # Guard against negated claims (e.g. "Alice did NOT finish the migration") + if _is_negated(text, match.start(), match.end()): + return conflicts + + claimed_person = match.group(1).strip() + claimed_task = match.group(2).strip() + + # Normalize task for comparison + task_lower = claimed_task.lower().replace(" ", "_").replace("-", "_") + + # Look for any KG triples about this task + for pred in _ATTRIBUTION_PREDICATES: + results = kg.query_relationship(pred) + for fact in results: + if not fact.get("current", False): + continue + fact_obj = fact["object"].lower().replace(" ", "_").replace("-", "_") + # Check if this fact is about the same task + if task_lower in fact_obj or fact_obj in task_lower: + fact_person = fact["subject"] + if fact_person.lower() != claimed_person.lower(): + conflicts.append( + Conflict( + severity="RED", + entity=claimed_person, + field="attribution", + message=( + f"attribution conflict — {fact_person} is " + f"{pred.replace('_', ' ')} {fact['object']}, " + f"not {claimed_person}" + ), + kg_fact=fact, + ) + ) + return conflicts + + +def _check_tenure(text: str, kg) -> List[Conflict]: + """Check if a tenure claim conflicts with KG start dates.""" + conflicts = [] + match = _TENURE_PATTERN.search(text) + if not match: + return conflicts + + claimed_person = match.group(1).strip() + claimed_years = int(match.group(2)) + + # Look for employment/joining triples + results = kg.query_entity(claimed_person, direction="outgoing") + for fact in results: + if not fact.get("current", False): + continue + pred = fact["predicate"] + if pred in ("works_at", "joined", "started_at", "employed_by"): + valid_from = fact.get("valid_from") + if valid_from: + try: + start_year = int(valid_from[:4]) + actual_years = date.today().year - start_year + if abs(actual_years - claimed_years) >= 1: + conflicts.append( + Conflict( + severity="YELLOW", + entity=claimed_person, + field="tenure", + message=( + f"tenure mismatch — records show " + f"{actual_years} years (started {valid_from}), " + f"not {claimed_years}" + ), + kg_fact=fact, + ) + ) + except (ValueError, IndexError): + pass + return conflicts + + +def _check_role(text: str, kg) -> List[Conflict]: + """Check if a role claim conflicts with KG role facts.""" + conflicts = [] + match = _ROLE_PATTERN.search(text) + if not match: + return conflicts + + if _is_negated(text, match.start(), match.end()): + return conflicts + + claimed_person = match.group(1).strip() + claimed_role = match.group(2).strip().lower() + + results = kg.query_entity(claimed_person, direction="outgoing") + for fact in results: + if not fact.get("current", False): + continue + if fact["predicate"] in ("has_role", "role", "position", "title"): + kg_role = fact["object"].lower() + if kg_role != claimed_role and claimed_role not in kg_role and kg_role not in claimed_role: + conflicts.append( + Conflict( + severity="RED", + entity=claimed_person, + field="role", + message=( + f"role conflict — records show {fact['object']}, " + f"not {claimed_role}" + ), + kg_fact=fact, + ) + ) + return conflicts + + +def _check_relationship(text: str, kg) -> List[Conflict]: + """Check if a relationship claim conflicts with KG relationship facts.""" + conflicts = [] + match = _RELATIONSHIP_PATTERN.search(text) + if not match: + return conflicts + + if _is_negated(text, match.start(), match.end()): + return conflicts + + person_a = match.group(1).strip() + person_b = match.group(2).strip() + claimed_rel = match.group(3).strip().lower() + + expected_predicates = _RELATIONSHIP_PAIRS.get(claimed_rel, ()) + if not expected_predicates: + return conflicts + + # Check outgoing from person_a + results_a = kg.query_entity(person_a, direction="both") + for fact in results_a: + if not fact.get("current", False): + continue + + # Is there a relationship between these two people? + other = fact["object"] if fact["subject"].lower() == person_a.lower() else fact["subject"] + if other.lower() != person_b.lower(): + continue + + # There IS a relationship — does it match the claimed one? + if fact["predicate"] not in expected_predicates: + conflicts.append( + Conflict( + severity="RED", + entity=person_a, + field="relationship", + message=( + f"relationship conflict — {person_a} is " + f"{fact['predicate'].replace('_', ' ')} {person_b}, " + f"not {claimed_rel}" + ), + kg_fact=fact, + ) + ) + return conflicts + + +def _extract_entity_names(text: str) -> List[str]: + """Extract capitalized names from text (simple heuristic).""" + # Find capitalized words that aren't at sentence start + words = text.split() + names = [] + for i, word in enumerate(words): + clean = re.sub(r"[^a-zA-Z]", "", word) + if ( + len(clean) >= 2 + and clean[0].isupper() + and clean[1:].islower() + and clean.lower() not in _COMMON_WORDS + ): + names.append(clean) + return list(dict.fromkeys(names)) # dedupe, preserve order + + +_COMMON_WORDS = { + "the", "this", "that", "these", "those", "here", "there", + "has", "have", "had", "was", "were", "been", "being", + "not", "but", "and", "for", "with", "from", "about", + "into", "over", "after", "before", "between", "under", + "again", "further", "then", "once", "also", "just", + "only", "very", "much", "many", "some", "any", "each", + "every", "both", "few", "more", "most", "other", "such", + "than", "too", "very", "can", "will", "should", "would", + "could", "may", "might", "shall", "must", "need", "now", + "new", "old", "big", "small", "long", "short", "high", + "low", "great", "good", "bad", "right", "wrong", "true", + "false", "yes", "all", "own", "same", "different", +} + + +# ── Public API ─────────────────────────────────────────────────────── + + +def check_assertion(text: str, kg) -> CheckResult: + """Check a text assertion against the knowledge graph for contradictions. + + Runs all checkers (attribution, tenure, role, relationship) and + returns the highest severity found. + + Args: + text: Natural language assertion to check. + kg: A KnowledgeGraph instance to query against. + + Returns: + CheckResult with severity, conflicts, and entities checked. + """ + all_conflicts = [] + all_conflicts.extend(_check_attribution(text, kg)) + all_conflicts.extend(_check_tenure(text, kg)) + all_conflicts.extend(_check_role(text, kg)) + all_conflicts.extend(_check_relationship(text, kg)) + + entities = _extract_entity_names(text) + + # Determine overall severity + if any(c.severity == "RED" for c in all_conflicts): + severity = "RED" + elif any(c.severity == "YELLOW" for c in all_conflicts): + severity = "YELLOW" + else: + severity = "GREEN" + + return CheckResult( + severity=severity, + text=text, + conflicts=all_conflicts, + entities_checked=entities, + ) diff --git a/mempalace/mcp_server.py b/mempalace/mcp_server.py index bffd3b2f2d..4f5f307bc7 100644 --- a/mempalace/mcp_server.py +++ b/mempalace/mcp_server.py @@ -471,6 +471,14 @@ def tool_kg_stats(): return _kg.stats() +def tool_check_facts(text: str): + """Check a statement for contradictions against the knowledge graph.""" + from mempalace.fact_checker import check_assertion + + result = check_assertion(text, _kg) + return result.to_dict() + + # ==================== AGENT DIARY ==================== @@ -699,6 +707,20 @@ def tool_diary_read(agent_name: str, last_n: int = 10): "input_schema": {"type": "object", "properties": {}}, "handler": tool_kg_stats, }, + "mempalace_check_facts": { + "description": "Check a statement for contradictions against the knowledge graph. Returns RED (direct conflict), YELLOW (numeric mismatch), or GREEN (no conflict). Use before recording claims to catch errors.", + "input_schema": { + "type": "object", + "properties": { + "text": { + "type": "string", + "description": "Natural language assertion to fact-check (e.g. 'Soren finished the auth migration')", + }, + }, + "required": ["text"], + }, + "handler": tool_check_facts, + }, "mempalace_traverse": { "description": "Walk the palace graph from a room. Shows connected ideas across wings — the tunnels. Like following a thread through the palace: start at 'chromadb-setup' in wing_code, discover it connects to wing_myproject (planning) and wing_user (feelings about it).", "input_schema": { diff --git a/tests/test_fact_checker.py b/tests/test_fact_checker.py new file mode 100644 index 0000000000..030368bedd --- /dev/null +++ b/tests/test_fact_checker.py @@ -0,0 +1,225 @@ +""" +test_fact_checker.py — Tests for rule-based contradiction detection. + +Covers: attribution conflicts, tenure mismatches, role conflicts, +relationship conflicts, no-conflict cases, unknown entities, +and expired fact handling. +""" + +from mempalace.fact_checker import check_assertion + + +# ── Attribution conflicts ──────────────────────────────────────────── + + +class TestAttributionConflicts: + def test_detects_wrong_person(self, kg): + """If KG says Maya is assigned to auth migration, claiming Soren did it is RED.""" + kg.add_entity("Maya", entity_type="person") + kg.add_entity("auth migration", entity_type="project") + kg.add_triple("Maya", "assigned_to", "auth migration") + + result = check_assertion("Soren finished the auth migration", kg) + assert result.severity == "RED" + assert len(result.conflicts) >= 1 + assert result.conflicts[0].field == "attribution" + assert "Maya" in result.conflicts[0].message + + def test_correct_attribution_is_green(self, kg): + """If KG says Maya is assigned and we claim Maya did it — no conflict.""" + kg.add_entity("Maya", entity_type="person") + kg.add_entity("auth migration", entity_type="project") + kg.add_triple("Maya", "assigned_to", "auth migration") + + result = check_assertion("Maya finished the auth migration", kg) + assert result.severity == "GREEN" + assert len(result.conflicts) == 0 + + def test_negated_attribution_is_green(self, kg): + """Negated claims should not trigger conflicts.""" + kg.add_entity("Maya", entity_type="person") + kg.add_entity("auth migration", entity_type="project") + kg.add_triple("Maya", "assigned_to", "auth migration") + + result = check_assertion("Soren did NOT finish the auth migration", kg) + assert result.severity == "GREEN" + + def test_no_longer_negation(self, kg): + """'no longer' should be recognized as negation.""" + kg.add_entity("Maya", entity_type="person") + kg.add_entity("auth migration", entity_type="project") + kg.add_triple("Maya", "assigned_to", "auth migration") + + result = check_assertion("Soren no longer finished the auth migration", kg) + assert result.severity == "GREEN" + + def test_no_kg_data_is_green(self, kg): + """If KG has no info about the task, no conflict can be detected.""" + result = check_assertion("Soren finished the auth migration", kg) + assert result.severity == "GREEN" + + +# ── Tenure mismatches ──────────────────────────────────────────────── + + +class TestTenureMismatches: + def test_detects_wrong_tenure(self, kg): + """If KG shows Kai started in 2020, claiming 2 years is wrong in 2026.""" + kg.add_entity("Kai", entity_type="person") + kg.add_entity("Acme Corp", entity_type="company") + kg.add_triple("Kai", "works_at", "Acme Corp", valid_from="2020-03-01") + + result = check_assertion("Kai has been here 2 years", kg) + assert result.severity == "YELLOW" + assert len(result.conflicts) >= 1 + assert result.conflicts[0].field == "tenure" + assert "2020" in result.conflicts[0].message + + def test_correct_tenure_is_green(self, kg): + """If KG start date matches the claimed tenure, no conflict.""" + current_year = 2026 # test assumes current year + start_year = current_year - 3 + kg.add_entity("Kai", entity_type="person") + kg.add_entity("Acme Corp", entity_type="company") + kg.add_triple("Kai", "works_at", "Acme Corp", valid_from=f"{start_year}-01-01") + + # Claiming ~3 years should be close enough (within 1 year tolerance) + result = check_assertion("Kai has been here 3 years", kg) + assert result.severity == "GREEN" + + def test_no_start_date_is_green(self, kg): + """If KG has the employment but no start date, can't check tenure.""" + kg.add_entity("Kai", entity_type="person") + kg.add_entity("Acme Corp", entity_type="company") + kg.add_triple("Kai", "works_at", "Acme Corp") + + result = check_assertion("Kai has been here 5 years", kg) + assert result.severity == "GREEN" + + +# ── Role conflicts ─────────────────────────────────────────────────── + + +class TestRoleConflicts: + def test_detects_wrong_role(self, kg): + """If KG says Alice is an engineer, claiming she's a designer is RED.""" + kg.add_entity("Alice", entity_type="person") + kg.add_entity("engineer", entity_type="role") + kg.add_triple("Alice", "has_role", "engineer") + + result = check_assertion("Alice is a designer at the company", kg) + assert result.severity == "RED" + assert result.conflicts[0].field == "role" + + def test_correct_role_is_green(self, kg): + """If KG says Alice is an engineer and we claim the same — green.""" + kg.add_entity("Alice", entity_type="person") + kg.add_entity("engineer", entity_type="role") + kg.add_triple("Alice", "has_role", "engineer") + + result = check_assertion("Alice is an engineer at the company", kg) + assert result.severity == "GREEN" + + def test_partial_role_match_is_green(self, kg): + """If KG says 'senior engineer' and claim says 'engineer' — no conflict.""" + kg.add_entity("Alice", entity_type="person") + kg.add_entity("senior engineer", entity_type="role") + kg.add_triple("Alice", "has_role", "senior engineer") + + result = check_assertion("Alice is an engineer at the company", kg) + assert result.severity == "GREEN" + + +# ── Relationship conflicts ─────────────────────────────────────────── + + +class TestRelationshipConflicts: + def test_detects_wrong_relationship(self, seeded_kg): + """seeded_kg has Alice parent_of Max. Claiming Max is Alice's partner is RED.""" + result = check_assertion("Max is Alice's partner", seeded_kg) + assert result.severity == "RED" + assert result.conflicts[0].field == "relationship" + + def test_correct_relationship_is_green(self, seeded_kg): + """seeded_kg has Alice parent_of Max. Claiming Max is Alice's child is green.""" + result = check_assertion("Max is Alice's child", seeded_kg) + assert result.severity == "GREEN" + + def test_no_relationship_data_is_green(self, kg): + """If KG has no relationship between two people, no conflict.""" + kg.add_entity("Zara", entity_type="person") + kg.add_entity("Liam", entity_type="person") + + result = check_assertion("Zara is Liam's sister", kg) + assert result.severity == "GREEN" + + +# ── Expired facts ──────────────────────────────────────────────────── + + +class TestExpiredFacts: + def test_expired_attribution_not_flagged(self, kg): + """Expired KG facts should not trigger conflicts.""" + kg.add_entity("Maya", entity_type="person") + kg.add_entity("auth migration", entity_type="project") + kg.add_triple( + "Maya", "assigned_to", "auth migration", + valid_from="2025-01-01", valid_to="2025-06-01", + ) + + result = check_assertion("Soren finished the auth migration", kg) + assert result.severity == "GREEN" + + def test_expired_employment_not_flagged(self, seeded_kg): + """Alice's old job at Acme Corp (ended 2024-12-31) shouldn't conflict.""" + # seeded_kg has: Alice works_at Acme Corp (valid_to=2024-12-31) + # Alice works_at NewCo (current) + result = check_assertion("Alice has been here 1 years", seeded_kg) + # Should only check current employment (NewCo, started 2025-01-01) + # 2026 - 2025 = 1 year, claimed 1 year → GREEN + assert result.severity == "GREEN" + + +# ── Result structure ───────────────────────────────────────────────── + + +class TestCheckResultStructure: + def test_to_dict(self, kg): + result = check_assertion("Hello world", kg) + d = result.to_dict() + assert "severity" in d + assert "text" in d + assert "conflicts" in d + assert "entities_checked" in d + assert isinstance(d["conflicts"], list) + + def test_entities_extracted(self, kg): + result = check_assertion("Alice told Bob about the new plan", kg) + assert "Alice" in result.entities_checked + assert "Bob" in result.entities_checked + + def test_empty_text(self, kg): + result = check_assertion("", kg) + assert result.severity == "GREEN" + assert len(result.conflicts) == 0 + + +# ── MCP integration ────────────────────────────────────────────────── + + +class TestMcpIntegration: + def test_tool_check_facts_returns_dict(self, kg): + """Verify the result format matches what MCP tools expect.""" + kg.add_entity("Maya", entity_type="person") + kg.add_triple("Maya", "assigned_to", "auth migration") + + result = check_assertion("Soren finished the auth migration", kg) + d = result.to_dict() + + assert d["severity"] == "RED" + assert len(d["conflicts"]) >= 1 + conflict = d["conflicts"][0] + assert "severity" in conflict + assert "entity" in conflict + assert "field" in conflict + assert "message" in conflict