"""GapDetector + HypothesisEngine — fill in missing relations. Third stage of the HMS cognition engine. Two stages in one module because they're tightly coupled: GapDetector — for each entity, compare its set of relations to its peers' sets. Missing relations that peers have are gaps. For example: Alice has {works_at, lives_in, has_skill} Bob has {works_at, lives_in, has_skill, age} → Alice is missing an `age` fact — gap. HypothesisEngine — for each gap, propose a filler. The filler can come from: (a) Majority vote: what value do peers of the same abstraction have? (b) Structural transitivity: if Alice's father is Bob and Bob's father is Charles, the engine hypothesizes Alice's grandfather is Charles via the father→father chain. (c) Hopfield cleanup: if a VSA palace is available, query the superposition of peers' values for that relation. Hypotheses are written as facts with: is_derived=1 confidence < 0.5 provenance.kind = "hypothesis" provenance.source = the basis of the guess The HYPOTHESIZED_BY edge links the supporting fact(s) to the new hypothesis fact, so audits can trace the reasoning chain. """ from __future__ import annotations import json import sqlite3 from collections import defaultdict from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any from cortexm.cognition.abstraction import Abstraction from cortexm.cognition.scanner import ScanResult from cortexm.trace.fact import deterministic_fact_id, make_fact from cortexm.trace.store import TraceStore from cortexm.util import iso, new_id # Edge kind used to wire supporting facts to hypotheses HYPOTHESIZED_BY = "HYPOTHESIZED_BY" @dataclass class Gap: """A missing relation for an entity.""" subject: str missing_relation: str peer_count: int = 0 # how many peers have this relation peer_values: list[str] = field(default_factory=list) basis: str = "" # "majority" | "structural" | "cleanup" @dataclass class Hypothesis: """A proposed filler for a gap.""" subject: str relation: str proposed_value: str confidence: float = 0.0 basis: str = "" # which strategy proposed this supporting_facts: list[str] = field(default_factory=list) fact_id: str = "" @dataclass class GapResult: gaps: list[Gap] = field(default_factory=list) n_subjects_compared: int = 0 duration_ms: float = 0.0 @dataclass class HypothesisResult: hypotheses: list[Hypothesis] = field(default_factory=list) facts_added: int = 0 duration_ms: float = 0.0 class GapDetector: """Finds missing relations by comparing entity profiles to peers.""" def __init__(self, store: TraceStore, min_peers_with_relation: int = 2) -> None: self.store = store self.min_peers_with_relation = min_peers_with_relation def run(self, scan: ScanResult, user_id: str | None = None) -> GapResult: """Compare each subject's relation set to its peers' sets.""" import time t0 = time.perf_counter() # build per-subject relation sets from the trace where = "is_active=1 AND quarantined=0" args: tuple = () if user_id is not None: where += " AND user_id=?" args = (user_id,) rows = self.store.conn.execute( f"SELECT subject, relation, value FROM facts WHERE {where}", args).fetchall() subj_rels: dict[str, set[str]] = defaultdict(set) subj_vals: dict[str, dict[str, str]] = defaultdict(dict) rel_subjects: dict[str, set[str]] = defaultdict(set) rel_values: dict[str, list[str]] = defaultdict(list) for r in rows: s, rel, v = r[0], r[1], r[2] subj_rels[s].add(rel) subj_vals[s][rel] = v rel_subjects[rel].add(s) rel_values[rel].append(v) # for each relation that exists in the trace, find subjects # that DON'T have it but have at least one co-occurring relation # with peers that DO have it. gaps: list[Gap] = [] all_subjects = set(subj_rels.keys()) for rel, peers in rel_subjects.items(): if len(peers) < self.min_peers_with_relation: continue missing_subjects = all_subjects - peers for s in missing_subjects: # check overlap with peers — at least one shared relation my_rels = subj_rels[s] peer_rels_union = set() for p in peers: peer_rels_union |= subj_rels[p] overlap = my_rels & peer_rels_union if len(overlap) < 1: continue # not a peer of these — different category gaps.append(Gap( subject=s, missing_relation=rel, peer_count=len(peers), peer_values=sorted(set(rel_values[rel]))[:5], basis="majority")) # structural gaps: for relations that form chain signatures # (e.g. father → father = grandfather), check each subject that # has the first hop — does it have the result of the second hop # recorded? If not, that's a structural gap. for p in scan.patterns: if p.kind != "relation_pair": continue chain_rel = p.payload["relation"] for ex in p.payload.get("examples", []): # ex = (start, mid, end). The subject `start` has # chain_rel=start→mid, mid→end. Hypothesize that # start has a chain_rel_2 value of end. # This is reported as a structural gap so the # HypothesisEngine can propose (start, chain_rel*2, end). if len(ex) != 3: continue start, mid, end = ex # check if start already has the inferred composite # relation — we model the composite as the chain # relation with the value being the chain endpoint. # Specifically: a hypothesis fact (start, chain_rel, # end) is misleading because start already has # chain_rel=mid. We use a synthetic relation name. composite_rel = f"{chain_rel}*{chain_rel}" existing = self.store.conn.execute( "SELECT 1 FROM facts WHERE subject=? AND relation=? " "AND value=? AND is_active=1 LIMIT 1", (start, composite_rel, end)).fetchone() if existing: continue # already known — not a gap gaps.append(Gap( subject=start, missing_relation=composite_rel, peer_count=p.support, peer_values=[end], basis="structural")) return GapResult( gaps=gaps, n_subjects_compared=len(all_subjects), duration_ms=(time.perf_counter() - t0) * 1000.0) class HypothesisEngine: """Proposes fillers for gaps via three strategies.""" MAX_HYPOTHESES_PER_GAP: int = 1 MAX_HYPOTHESIS_CONFIDENCE: float = 0.45 def __init__(self, store: TraceStore, palace=None, max_confidence: float | None = None) -> None: self.store = store self.palace = palace # for Hopfield cleanup AND vector indexing self.max_conf = max_confidence or self.MAX_HYPOTHESIS_CONFIDENCE def run(self, gaps: list[Gap], *, dry_run: bool = False, commit_id: str | None = None, user_id: str | None = None, palace=None) -> HypothesisResult: """Propose and write hypothesis facts for each gap.""" import time t0 = time.perf_counter() hypotheses: list[Hypothesis] = [] facts_added = 0 ts = iso(_now()) if not dry_run else "" for gap in gaps: proposed = self._propose(gap, user_id) if not proposed: continue hypotheses.append(proposed) if dry_run: continue # write the hypothesis as a derived fact # id excludes the derivation timestamp — see the matching # note in abstraction.py (re-runs skip) fid = deterministic_fact_id( user_id=user_id or "default", subject=proposed.subject, relation=proposed.relation, value=proposed.proposed_value) try: self.store.conn.execute( "INSERT INTO facts " "(id, subject, relation, value, valid_from, tx_from, " " confidence, user_id, memory_type, is_derived, " " is_active, birth_commit, provenance) " "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", (fid, proposed.subject, proposed.relation, proposed.proposed_value, ts, ts, proposed.confidence, user_id or "default", "long_term", 1, 1, commit_id, json.dumps({ "kind": "hypothesis", "basis": proposed.basis, "supporting_facts": proposed.supporting_facts, "peer_count": gap.peer_count, "peer_values_sample": gap.peer_values, "generated_by": "cognition.hypothesis", }))) # Add vector to palace for retrieval use_palace = palace or self.palace if use_palace is not None: fact = make_fact( subject=proposed.subject, relation=proposed.relation, value=proposed.proposed_value, user_id=user_id or "default", confidence=proposed.confidence, memory_type="long_term", is_derived=True, is_active=True, now=datetime.now(timezone.utc), provenance={"kind": "hypothesis", "basis": proposed.basis, "supporting_facts": proposed.supporting_facts, "peer_count": gap.peer_count, "peer_values_sample": gap.peer_values, "generated_by": "cognition.hypothesis"}) fact.id = fid use_palace.add(fid, use_palace.encode_fact(fact)) except sqlite3.IntegrityError: continue # already derived (deterministic id) — idempotent # wire HYPOTHESIZED_BY edges from each supporting fact for support_fid in proposed.supporting_facts: if support_fid: try: self.store.add_edge( fid, support_fid, HYPOTHESIZED_BY, {"basis": proposed.basis}) except Exception: pass proposed.fact_id = fid facts_added += 1 if not dry_run and commit_id and facts_added: self.store.update_commit_n_facts(commit_id, facts_added) return HypothesisResult( hypotheses=hypotheses, facts_added=facts_added, duration_ms=(time.perf_counter() - t0) * 1000.0) # ---------- strategies ------------------------------------------- def _propose(self, gap: Gap, user_id: str | None) -> Hypothesis | None: # structural: peer_values are exact (end of chain) if gap.basis == "structural" and gap.peer_values: value = gap.peer_values[0] supporting = self._lookup_supporting_facts( gap.subject, gap.missing_relation, value, user_id) return Hypothesis( subject=gap.subject, relation=gap.missing_relation, proposed_value=value, confidence=min(self.max_conf, 0.20 + 0.05 * gap.peer_count), basis="structural", supporting_facts=supporting, ) # majority: most common value among peers if gap.basis == "majority" and gap.peer_values: # majority among peers — for categorical relations like # `has_skill:rust` or `lives_in:toronto`, the most common # peer value is a reasonable guess. For unique values # like `works_at` (each person works at one company), the # majority strategy is weaker — we still emit a hypothesis # but at much lower confidence. counter: dict[str, int] = defaultdict(int) where = "is_active=1 AND quarantined=0 AND relation=?" args: tuple = (gap.missing_relation,) if user_id is not None: where += " AND user_id=?" args = (gap.missing_relation, user_id) rows = self.store.conn.execute( f"SELECT value FROM facts WHERE {where}", args).fetchall() for r in rows: counter[r[0]] += 1 if not counter: return None most_common = max(counter.items(), key=lambda x: x[1]) value = most_common[0] n_peers_with_value = most_common[1] total_peers = sum(counter.values()) conf = min(self.max_conf, 0.05 + 0.05 * (n_peers_with_value / max(1, total_peers))) supporting = self._lookup_supporting_facts( gap.subject, gap.missing_relation, value, user_id) return Hypothesis( subject=gap.subject, relation=gap.missing_relation, proposed_value=value, confidence=conf, basis="majority", supporting_facts=supporting, ) return None def _lookup_supporting_facts(self, subject: str, relation: str, value: str, user_id: str | None) -> list[str]: """Look up fact IDs that support a hypothesis. For structural hypotheses (father→father), supporting facts are the two hops (Alice father Bob, Bob father Charles). """ ids: list[str] = [] if relation.endswith("*" + relation.split("*")[0]): # composite chain — look up the two hops base_rel = relation.split("*")[0] cur = subject for _ in range(2): rows = self.store.conn.execute( "SELECT id, value FROM facts WHERE subject=? " "AND relation=? AND is_active=1 " "AND quarantined=0 LIMIT 1", (cur, base_rel)).fetchall() if not rows: break ids.append(rows[0][0]) cur = rows[0][1] if cur == value: break else: rows = self.store.conn.execute( "SELECT id FROM facts WHERE relation=? AND value=? " "AND is_active=1 AND quarantined=0 LIMIT 5", (relation, value)).fetchall() ids = [r[0] for r in rows] return ids def _now(): from datetime import datetime, timezone return datetime.now(timezone.utc) __all__ = [ "GapDetector", "HypothesisEngine", "Gap", "Hypothesis", "GapResult", "HypothesisResult", "HYPOTHESIZED_BY", ]