"""CognitionEngine — orchestrates the 5-stage self-organization pipeline. Triggered from `cortexm consolidate` (or `Memory.consolidate()`) so the engine is deterministic, auditable, and never produces surprise writes in the foreground path. Each pass: 1. PatternScanner.run() — surface structural regularities 2. AbstractionEngine.run() — build prototype categories 3. GapDetector.run() — find missing relations 4. HypothesisEngine.run() — propose fillers for gaps 5. AnalogyDetector.run() — find cross-domain analogies Output is committed as derived facts with confidence < 0.5 and tagged is_derived=1, so they don't pollute the active fact count and don't fire in retrieval unless explicitly promoted (via `Memory.promote_hypothesis(fact_id)`). The HYPOTHESIZED_BY edge kind is registered here so `cortexm.trace.edges.ALL_KINDS` includes it. PROMOTED_FROM is the inverse — when a hypothesis is confirmed by user input, the new fact points back to its hypothesis origin. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any from cortexm.cognition.scanner import PatternScanner, ScanResult from cortexm.cognition.abstraction import AbstractionEngine, AbstractionResult from cortexm.cognition.gaps import ( GapDetector, HypothesisEngine, GapResult, HypothesisResult, HYPOTHESIZED_BY) from cortexm.cognition.analogy import ( AnalogyDetector, AnalogyResult, ANALOGOUS_TO) from cortexm.trace.store import TraceStore # Edge kinds produced by the cognition engine. These are added to the # trace.edges vocabulary. We register them lazily on first cognition # pass so a Trace that never ran cognition doesn't see these kinds in # queries (avoiding schema pollution). PROMOTED_FROM = "PROMOTED_FROM" # user-confirmed fact -> hypothesis origin ABSTRACTS = "ABSTRACTS" # abstraction -> member INSTANTIATES = "INSTANTIATES" # member -> abstraction @dataclass class CognitionReport: """Full report of one cognition pass.""" scan: dict = field(default_factory=dict) abstraction: dict = field(default_factory=dict) gaps: dict = field(default_factory=dict) hypotheses: dict = field(default_factory=dict) analogies: dict = field(default_factory=dict) commit_id: str | None = None dry_run: bool = False total_derived_facts: int = 0 duration_ms: float = 0.0 class CognitionEngine: """Orchestrates the 5-stage self-organization pipeline.""" def __init__(self, store: TraceStore, palace=None) -> None: self.store = store self.palace = palace self.scanner = PatternScanner(store) self.abstraction = AbstractionEngine(store) self.gap_detector = GapDetector(store) # pass palace for Hopfield cleanup path (currently not used # by the simple strategies but the API accepts it) self.hypothesis = HypothesisEngine(store, palace=palace) self.analogy = AnalogyDetector(store) def run(self, *, dry_run: bool = False, user_id: str | None = None) -> CognitionReport: """Run the full 5-stage pipeline. All writes happen inside a single commit if not dry_run. Returns a CognitionReport with per-stage stats. """ import time t0 = time.perf_counter() # if not dry_run, we expect the caller to have already started # a batch + commit (consolidate() does this). We accept an # optional commit_id from the caller via store._cognition_commit # (stored as a session-level hint). If not present, we create # our own commit. own_commit = False commit_id = None if not dry_run: try: self.store.begin_batch() commit_id = self.store.create_commit( "cognition: scan+abstract+gap+hypothesis+analogy", n_facts=0) own_commit = True except Exception: # caller already in a batch — use their commit commit_id = getattr(self.store, "_active_commit_id", None) report = CognitionReport(dry_run=dry_run, commit_id=commit_id) # 1. PatternScanner scan = self.scanner.run(user_id=user_id) report.scan = { "patterns": len(scan.patterns), "n_facts_scanned": scan.n_facts_scanned, "n_relations": scan.n_relations, "n_subjects": scan.n_subjects, "n_values": scan.n_values, "duration_ms": round(scan.duration_ms, 2), "by_kind": _count_by_kind(scan.patterns), } # 2. AbstractionEngine ab = self.abstraction.run(scan, dry_run=dry_run, commit_id=commit_id, user_id=user_id, palace=self.palace) report.abstraction = { "abstractions": len(ab.abstractions), "membership_edges_added": ab.membership_edges_added, "duration_ms": round(ab.duration_ms, 2), } # 3. GapDetector gaps = self.gap_detector.run(scan, user_id=user_id) report.gaps = { "gaps": len(gaps.gaps), "n_subjects_compared": gaps.n_subjects_compared, "duration_ms": round(gaps.duration_ms, 2), "by_basis": _count_gaps_by_basis(gaps.gaps), } # 4. HypothesisEngine hyp = self.hypothesis.run(gaps.gaps, dry_run=dry_run, commit_id=commit_id, user_id=user_id, palace=self.palace) report.hypotheses = { "hypotheses": len(hyp.hypotheses), "facts_added": hyp.facts_added, "duration_ms": round(hyp.duration_ms, 2), } # 5. AnalogyDetector ana = self.analogy.run(scan, dry_run=dry_run, commit_id=commit_id, user_id=user_id, palace=self.palace) report.analogies = { "analogies": len(ana.analogies), "edges_added": ana.edges_added, "duration_ms": round(ana.duration_ms, 2), } report.total_derived_facts = ( ab.membership_edges_added + hyp.facts_added + ana.edges_added ) report.duration_ms = (time.perf_counter() - t0) * 1000.0 if not dry_run and own_commit: # update commit n_facts to the actual count we wrote try: self.store.update_commit_n_facts( commit_id, report.total_derived_facts) except Exception: pass try: self.store.end_batch() except Exception: pass return report def run_cognition_pass(store: TraceStore, palace=None, *, dry_run: bool = False, user_id: str | None = None) -> CognitionReport: """Convenience function: one-shot cognition pass on a TraceStore.""" return CognitionEngine(store, palace=palace).run( dry_run=dry_run, user_id=user_id) def _count_by_kind(patterns) -> dict[str, int]: out: dict[str, int] = {} for p in patterns: out[p.kind] = out.get(p.kind, 0) + 1 return out def _count_gaps_by_basis(gaps) -> dict[str, int]: out: dict[str, int] = {} for g in gaps: out[g.basis] = out.get(g.basis, 0) + 1 return out __all__ = [ "CognitionEngine", "CognitionReport", "run_cognition_pass", "HYPOTHESIZED_BY", "PROMOTED_FROM", "ABSTRACTS", "INSTANTIATES", "ANALOGOUS_TO", ]