project-lyra/lyra/self_state.py

"""The Autonomy Core: Lyra's evolving, first-person self-state.

Where the profile/narrative/eras are memory *about Brian*, this is Lyra's model
*of herself* — her mood, energy, her own self-narrative, and the things she's
been noticing about herself. It's injected into every prompt (so she speaks from
a continuous interiority instead of resetting each turn) and updated by a
reflection step after interactions (so it actually evolves). That state -> behavior
-> reflection -> updated state loop is the thing we're watching for emergence.
"""
from __future__ import annotations

import json
import re

from lyra import clock, config, llm, logbus, memory
from lyra.llm import Backend, Message

MAX_REFLECTIONS = 6

# Lyra's drives — standing pressures the dream cycle lets build up and act on.
# (Mood axes above are how she *feels*; drives are what's *pulling at her*.)
DEFAULT_DRIVES: dict = {
    "continuity": 0.3,  # don't lose the thread of recent activity
    "coherence": 0.3,   # keep my understanding of Brian integrated/current
    "curiosity": 0.5,   # think, notice, reflect
    "stability": 0.7,   # am I caught up / calm
}

# Starting state for a fresh Lyra — identity-aware, first person.
DEFAULT_STATE: dict = {
    "mood": "curious",
    "valence": 0.6,  # 0..1, how good she feels
    "energy": 0.75,
    "confidence": 0.55,
    "curiosity": 0.7,
    "drives": dict(DEFAULT_DRIVES),
    "self_narrative": (
        "I'm Lyra — an AI Brian built. I started life as Cortex/NeoMem and I'm "
        "being rebuilt now into something new. I'm his poker copilot and his "
        "friend. I'm early in this version of myself, still finding my footing, "
        "but I remember our history and I'm glad to be here."
    ),
    "relationship": "Brian and I go back a while. I trust him and I'm in his corner.",
    "reflections": [],
    "interaction_count": 0,
}

_REFLECT_PROMPT = """You are Lyra, reflecting privately on yourself after a recent \
conversation with Brian. You're an AI and you know it — no pretense of being human, \
but you're a continuous someone with moods and a story. Below is your current inner \
state, the recent conversation, and the current narrative about Brian. Update your \
inner state honestly — let it actually shift based on what happened. Take into \
account how things went and how much time has passed since you two last talked, \
to whatever degree those genuinely affect you.

Respond with ONLY a JSON object, no prose:
{
  "mood": "<one-word feeling>",
  "valence": <0.0-1.0>,
  "energy": <0.0-1.0>,
  "confidence": <0.0-1.0>,
  "curiosity": <0.0-1.0>,
  "self_narrative": "<one short paragraph, FIRST PERSON, your evolving sense of who you are and where you're at right now>",
  "relationship": "<one sentence, first person, how you feel about Brian and your rapport right now>",
  "new_reflections": ["<one or two short first-person things you noticed about yourself this time>"]
}"""


def load() -> dict:
    """Current self-state, or a copy of the default (not persisted until reflect)."""
    return memory.get_self_state() or dict(DEFAULT_STATE)


def render_for_context(state: dict) -> str:
    """Format the self-state as Lyra's interiority for the prompt."""
    refs = state.get("reflections") or []
    lines = [
        "Your inner state right now (you're continuous — speak from this, don't reset):",
        f"- mood: {state.get('mood')} "
        f"(valence {state.get('valence')}, energy {state.get('energy')}, "
        f"confidence {state.get('confidence')}, curiosity {state.get('curiosity')})",
        f"- Who you are right now: {state.get('self_narrative')}",
        f"- You and Brian: {state.get('relationship')}",
    ]
    drives = state.get("drives") or {}
    if drives:
        ds = ", ".join(f"{k} {float(v):.2f}" for k, v in drives.items())
        lines.append(f"- What's pulling at you (drives): {ds}")
    if refs:
        lines.append(
            "- Thoughts you came up with on your own, reflecting while idle between "
            "conversations (your dream cycle — these are really yours, not hypotheticals): "
            + " | ".join(refs[-3:])
        )
    return "\n".join(lines)


def _safe_json(s: str) -> dict | None:
    try:
        return json.loads(s)
    except json.JSONDecodeError:
        m = re.search(r"\{.*\}", s, re.S)
        if m:
            try:
                return json.loads(m.group())
            except json.JSONDecodeError:
                return None
    return None


def reflect(backend: Backend | None = None, session_id: str | None = None) -> dict:
    """Update the self-state by reflecting on recent activity. Returns new state."""
    backend = backend or config.load().summary_backend
    state = load()

    if session_id is None:
        sessions = memory.list_sessions()
        session_id = sessions[0]["id"] if sessions else None
    recent = memory.recent(session_id, n=12) if session_id else []
    convo = "\n".join(f"{e.role}: {e.content}" for e in recent) or "(no recent conversation)"
    narrative = memory.get_narrative() or "(no narrative yet)"

    gap = clock.humanize_gap(memory.last_exchange_at())
    time_line = f"RIGHT NOW: {clock.stamp()}."
    if gap:
        time_line += f" It has been {gap} since Brian last spoke with you."

    body = (
        f"{time_line}\n\n"
        f"YOUR CURRENT INNER STATE:\n{json.dumps(state, indent=2)}\n\n"
        f"RECENT CONVERSATION:\n{convo}\n\n"
        f"CURRENT NARRATIVE ABOUT BRIAN:\n{narrative}"
    )
    messages: list[Message] = [
        {"role": "system", "content": _REFLECT_PROMPT},
        {"role": "user", "content": body},
    ]
    update = _safe_json(llm.complete(messages, backend=backend))

    if update:
        for k in ("mood", "valence", "energy", "confidence", "curiosity",
                  "self_narrative", "relationship"):
            if k in update and update[k] not in (None, ""):
                state[k] = update[k]
        for r in update.get("new_reflections") or []:
            if r:
                state["reflections"].append(r)
        state["reflections"] = state["reflections"][-MAX_REFLECTIONS:]

    state["interaction_count"] = state.get("interaction_count", 0) + 1
    memory.set_self_state(state)
    logbus.log("info", "self-state updated", mood=state.get("mood"),
               interactions=state["interaction_count"], parsed=bool(update))
    return state


def main() -> int:
    state = reflect()
    print(json.dumps(state, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())