""" Cognitive Evolution Log for Berna R7. Records every event in the model's cognitive lifecycle: - birth: new cell created when new pattern discovered - split: existing cell divides due to saturation - pattern: new knowledge signature detected - milestone: notable thresholds crossed Persistent (JSONL), queryable, narrative-generating, feedable to model. """ import json import hashlib from pathlib import Path from datetime import datetime, timezone from typing import List, Optional class CognitiveLog: def __init__(self, log_dir, model_uuid="mother-001"): self.log_dir = Path(log_dir) self.log_dir.mkdir(parents=True, exist_ok=True) self.model_uuid = model_uuid self.jsonl_path = self.log_dir / f"{model_uuid}_events.jsonl" self.narrative_path = self.log_dir / f"{model_uuid}_narrative.txt" self.events = [] self.MAX_EVENTS = 10000 self._load_existing() def _load_existing(self): if self.jsonl_path.exists(): with open(self.jsonl_path, "r", encoding="utf-8") as f: for line in f: try: self.events.append(json.loads(line)) except Exception: pass def record(self, event_type, **fields): event = { "timestamp": datetime.now(timezone.utc).isoformat(), "step": fields.pop("step", None), "type": event_type, "model_uuid": self.model_uuid, } event.update(fields) self.events.append(event) if len(self.events) > self.MAX_EVENTS: self.events = self.events[-self.MAX_EVENTS:] with open(self.jsonl_path, "a", encoding="utf-8") as f: f.write(json.dumps(event, ensure_ascii=False) + "\n") return event def record_birth(self, cell_id, parent_id, pattern_sig, step, layer_idx): return self.record("birth", step=step, cell_id=cell_id, parent_id=parent_id, pattern_sig=pattern_sig, layer_idx=layer_idx) def record_split(self, parent_id, child_a, child_b, S, step, layer_idx): return self.record("split", step=step, parent_id=parent_id, child_a=child_a, child_b=child_b, S=round(float(S), 4), layer_idx=layer_idx) def record_pattern(self, pattern_sig, cell_ids, step, count): return self.record("pattern", step=step, pattern_sig=pattern_sig, cell_ids=cell_ids, seen_count=count) def record_milestone(self, name, value, step): return self.record("milestone", step=step, name=name, value=value) def recent(self, n=20, event_type=None): if event_type is None: return self.events[-n:] return [e for e in self.events if e["type"] == event_type][-n:] def count_by_type(self): counts = {} for e in self.events: counts[e["type"]] = counts.get(e["type"], 0) + 1 return counts def to_narrative(self, max_events=50): events = self.events[-max_events:] lines = [] lines.append(f"=== Berna {self.model_uuid} — Cognitive Chronicle ===") lines.append(f"Total events: {len(self.events)}") lines.append(f"Breakdown: {self.count_by_type()}") lines.append("") for e in events: t = e["type"] if t == "birth": lines.append(f"[step {e['step']}] Born: {e['cell_id']} " f"(from {e.get('parent_id','seed')}) — " f"knows pattern {str(e.get('pattern_sig','?'))[:8]}") elif t == "split": lines.append(f"[step {e['step']}] Split: {e['parent_id']} → " f"{e['child_a']} + {e['child_b']} (S={e['S']})") elif t == "pattern": lines.append(f"[step {e['step']}] New pattern: " f"{str(e['pattern_sig'])[:8]} " f"(seen {e['seen_count']}×, matched " f"{len(e.get('cell_ids', []))} cells)") elif t == "milestone": lines.append(f"[step {e['step']}] Milestone: " f"{e['name']} = {e['value']}") return "\n".join(lines) def write_narrative(self, path=None): text = self.to_narrative() p = Path(path) if path else self.narrative_path p.write_text(text, encoding="utf-8") return p def sample_for_training(self, n=2): """Short self-narrative snippet to prepend to training samples.""" births = self.recent(n=n, event_type="birth") splits = self.recent(n=n, event_type="split") lines = ["I am Berna R7, a self-growing language model."] if births: ids = ", ".join(e["cell_id"] for e in births) lines.append(f"Recently I gave birth to new cells: {ids}.") if splits: ids = ", ".join(e["parent_id"] for e in splits) lines.append(f"I also split {ids} to refine my knowledge.") if not births and not splits: lines.append("I am learning language, mathematics, and code.") return " ".join(lines) def pattern_signature(activations, top_k=50): """Binary fingerprint of top-K activated neurons, hashed.""" import numpy as np a = np.abs(np.asarray(activations, dtype=np.float32)) mean_a = a.mean(axis=0) if mean_a.size == 0: return "empty" k = min(top_k, mean_a.size) top_idx = np.argsort(mean_a)[-k:] fp = np.zeros(mean_a.size, dtype=np.uint8) fp[top_idx] = 1 return hashlib.sha256(fp.tobytes()).hexdigest()[:16] def hamming_similarity(fp_a, fp_b): """Similarity ∈ [0,1] between two fingerprints (hex strings).""" import numpy as np try: a = np.frombuffer(bytes.fromhex(fp_a), dtype=np.uint8) b = np.frombuffer(bytes.fromhex(fp_b), dtype=np.uint8) except Exception: return 0.0 n = min(len(a), len(b)) if n == 0: return 0.0 matches = np.sum(a[:n] == b[:n]) return float(matches / n) if __name__ == "__main__": import numpy as np log = CognitiveLog("/tmp/berna_test_log") log.record_birth("c-001", "seed", "abc123", 1, 0) log.record_split("c-000", "c-000-a", "c-000-b", 0.87, 100, 0) log.record_milestone("first_split", 1, 100) print(log.to_narrative()) print() a = np.random.randn(32, 128).astype(np.float32) b = np.random.randn(32, 128).astype(np.float32) print("fp_a:", pattern_signature(a)) print("fp_b:", pattern_signature(b)) print("sim:", hamming_similarity(pattern_signature(a), pattern_signature(b)))