project-lyra/lyra/cognition.py

"""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)