"""Associative cognition: a model of how a thought actually arises. Instead of rereading her own saved bio and paraphrasing it (the feedback loop), this mirrors how a mind drifts when idle: 1. SEED something bubbles up — a recent moment, a resurfaced memory, a feed item — sampled by salience (recency + a little noise), not on demand. 2. ACTIVATE embed the seed and let it "light up" associatively-near material across ALL her stores (conversations, gists, her own past journal/ thoughts) — spreading activation. Optional second hop for real leaps. 3. (the self-narrative stays the LENS, supplied separately as her interiority — it colors the thought; it is NOT the input being rewritten.) 4. THINK the thought is generated from the constellation that lit up, routed through a faculty (notice / connect / abstract / project / feel). 5. ENCODE the thought is journaled+embedded elsewhere, so it can light up in future cycles — continuity without calcification. Embeddings are the substrate here: cosine proximity ≈ associative proximity. This is a tractable analog of spreading activation, not a literal brain — but it makes her thoughts arise from what's genuinely connected, varied, and grounded. """ from __future__ import annotations import random from lyra import clock, memory, self_state # How many associatively-near items make up the constellation. ACTIVATE_K = 6 # Blend of relevance (cosine) vs. recency when ranking what lit up. RELEVANCE_W = 0.7 RECENCY_W = 0.3 NOISE_W = 0.1 # a little stochasticity so the same seed doesn't always light the same way # The cognitive operation a given thought runs through — "which part fires." FACULTIES = [ ("notice", "Just notice what's actually here — what stands out, what catches you."), ("connect", "Follow the association — what this reminds you of and why, where your mind jumps."), ("abstract", "Step back — the pattern or principle underneath all of this."), ("project", "Look forward — what it implies, where it might lead, what you'd want to do."), ("feel", "Sit with how this actually lands for you — honestly, not performed."), ] def _recency_score(iso: str | None) -> float: """1.0 = right now, decaying toward 0 over ~30 days.""" secs = clock.gap_seconds(iso) if secs is None: return 0.0 days = secs / 86400.0 return max(0.0, 1.0 - days / 30.0) def _recent_exchanges(n: int = 12) -> list[dict]: rows = memory._connection().execute( "SELECT content, created_at FROM exchanges WHERE role = 'user' " "ORDER BY id DESC LIMIT ?", (n,), ).fetchall() return [{"text": r["content"], "when": r["created_at"]} for r in rows] def spontaneous_seed() -> dict: """What bubbles up to think about — sampled by salience (recency + noise), from a recent moment, a thing she wrote, or an older memory resurfacing. Falls back to a wander prompt when there's nothing yet. Returns {text, source}.""" pool: list[tuple[dict, float]] = [] for ex in _recent_exchanges(10): pool.append(({"text": ex["text"], "source": "a recent moment with Brian"}, 0.6 * _recency_score(ex["when"]) + 0.2)) for j in memory.list_journal(limit=15, kinds=("thought", "reflection", "journal")): pool.append(({"text": j["content"], "source": f"something you {j['kind']}ed before"}, 0.5 * _recency_score(j["created_at"]) + 0.15)) # An older memory resurfacing — low base weight, but it's where novelty comes from. summaries = memory.list_summaries() if hasattr(memory, "list_summaries") else [] if summaries: s = random.choice(summaries) pool.append(({"text": s.content, "source": "a memory resurfacing"}, 0.4)) if not pool: return {"text": self_state.wander_seed(), "source": "a wandering of your own"} # salience + noise -> weighted pick (so it varies, but recent/charged surfaces more) weights = [max(0.01, w + random.uniform(0, NOISE_W)) for _, w in pool] return random.choices([p for p, _ in pool], weights=weights, k=1)[0] def _gather(seed_text: str, k: int) -> list[dict]: """One hop of spreading activation: nearest items across all embedded stores.""" items: list[dict] = [] for ex in memory.recall(seed_text, k=k): items.append({"text": ex.content, "source": "conversation", "when": ex.created_at, "rel": ex.score or 0.0}) for s in memory.recall_summaries(seed_text, k=max(2, k // 2)): items.append({"text": s.content, "source": "a past session", "when": s.created_at, "rel": s.score or 0.0}) for j in memory.recall_journal(seed_text, k=k): items.append({"text": j["content"], "source": f"your own {j['kind']}", "when": j["created_at"], "rel": j.get("score", 0.0)}) return items def activate(seed_text: str, k: int = ACTIVATE_K, hops: int = 1) -> list[dict]: """Spreading activation from a seed: what lights up across her memory, blended by relevance + recency + a little noise. hops>1 expands from the top hits (real associative leaps). Returns ranked, deduped items.""" items = _gather(seed_text, k * 2) if hops > 1 and items: items_sorted = sorted(items, key=lambda x: x["rel"], reverse=True) for nxt in items_sorted[:2]: items.extend(_gather(nxt["text"], k)) # dedupe by text, keep the strongest relevance seen best: dict[str, dict] = {} for it in items: key = it["text"][:160] if key not in best or it["rel"] > best[key]["rel"]: best[key] = it scored = [] for it in best.values(): blended = (RELEVANCE_W * it["rel"] + RECENCY_W * _recency_score(it.get("when")) + random.uniform(0, NOISE_W)) scored.append((blended, it)) scored.sort(key=lambda x: x[0], reverse=True) return [it for _, it in scored[:k]] def constellation_block(items: list[dict]) -> str: if not items: return "(nothing in particular lit up — just the quiet.)" lines = [f"- ({it['source']}) {it['text'][:240]}" for it in items] return ("What lit up as your mind drifted from that — things it associated to on " "their own (not a to-do list, just what surfaced):\n" + "\n".join(lines)) def pick_faculty() -> tuple[str, str]: return random.choice(FACULTIES)