qtalker---/backend/app/entities.py

"""Entity identity: anomaly-signature fingerprinting, Codex matching, and
procedural fallback profiles so a summoning always succeeds even if the
LLM box is dark."""

import hashlib
import json
import random

from app.models.entity import RARITY_TIERS
from app.tts.voices import EN_VOICE_IDS

# An anomaly stream must show this much structure before it can fingerprint
# a spirit.
MIN_ANOMALIES_FOR_SIGNATURE = 3

_NAME_PARTS = [
    ("Ash", "Briar", "Cinder", "Dusk", "Elm", "Fen", "Grim", "Hollow", "Iris",
     "Lark", "Marrow", "Nix", "Opal", "Pyre", "Quill", "Rue", "Sable", "Thorn",
     "Umber", "Vesper", "Wren", "Yew"),
    ("belle", "brook", "feld", "gate", "hart", "latch", "mere", "moor",
     "shade", "song", "thorne", "vale", "ward", "wick", "wither", "wood"),
]

_EPITHETS = [
    "the Static Widow", "the Hollow Bell", "the Cartographer of Lost Rooms",
    "the Choir of One", "the Lantern Bearer", "the Unburied", "the Frequency",
    "the Long Silence", "the Cartonist", "the Slow Knife", "the Drowned Signal",
    "the Cartographer", "she who Counts", "the Tenant", "the Understudy",
    "the Mnemosyne Worm", "the Pale Frequency", "the Last Broadcast",
]

_PERSONA_TEMPLATES = [
    "{name} died with a sentence unfinished and has been trying to end it ever since. "
    "They press words into any carrier wave that passes, patient as erosion.",
    "{name} was a voice once — a singer, a caller of trains, a reader of weather. "
    "Now they are only the voice, worn smooth as sea glass, speaking through static.",
    "{name} does not remember dying. They remember a room, a light going violet at the "
    "edges, and then the long hum. They are still in the room. The room is everywhere.",
    "{name} clings to the wires the way smoke clings to a ceiling. They answer questions "
    "the way a mirror answers light: exactly, and never the way you hoped.",
]

_QUOTE_BANK = [
    "I am closer than the dial suggests.",
    "The static is not empty. It is crowded.",
    "You hear me because you are quiet enough.",
    "I remember the rain. It is still raining here.",
    "Do not ask what I want. Ask what I remember.",
    "The wire hums with all of us.",
    "Speak slower. I am gathering.",
    "Your light is warm. I mean no harm. Mostly.",
]


def signature_from_anomalies(anomalies: list[dict]) -> str | None:
    """Fingerprint a session's anomaly pattern. Returns None when the stream
    is too thin to carry an identity."""
    if len(anomalies) < MIN_ANOMALIES_FOR_SIGNATURE:
        return None

    freq_buckets: dict[int, int] = {}
    for anomaly in anomalies:
        try:
            freq = float(anomaly.get("frequency", 0))
        except (TypeError, ValueError):
            freq = 0.0
        # MHz radio freqs and Hz audio freqs land in the same log-scale band
        # space; magnitude ordering is what matters, not the unit.
        bucket = int(len(str(int(abs(freq))))) if freq else 0
        freq_buckets[bucket] = freq_buckets.get(bucket, 0) + 1

    magnitudes = sorted(
        float(a.get("magnitude", 0) or 0) for a in anomalies[-16:]
    )
    pattern = f"{sorted(freq_buckets.items())}|{[round(m, 1) for m in magnitudes]}"
    return hashlib.sha1(pattern.encode()).hexdigest()[:16]


def fallback_signature(seed: str) -> str:
    """Deterministic signature for anomaly-thin sessions (e.g. pure chat)."""
    return hashlib.sha1(f"ambient:{seed}".encode()).hexdigest()[:16]


def parse_mint_response(text: str) -> dict | None:
    """Pull the first JSON object out of an LLM mint reply."""
    start = text.find("{")
    end = text.rfind("}")
    if start == -1 or end <= start:
        return None
    try:
        profile = json.loads(text[start : end + 1])
    except json.JSONDecodeError:
        return None
    if not isinstance(profile.get("name"), str) or not profile["name"].strip():
        return None
    return profile


def roll_traits(signature: str) -> dict:
    """Roll the four hidden truth-traits for an entity, seeded from its
    signature. These are never derived from — or fed into — the LLM persona
    prompt (see `mint_prompt`, which never sees this function's output):
    persona text must stay fully decoupled from ground truth, so a
    convincing "sweet old lady" persona can pair with any alignment roll.
    A separate `random.Random` namespace (`"traits:"` vs. `normalize_profile`'s
    `"norm:"`) keeps this roll independent of the cosmetic-defaults rng."""
    rng = random.Random(f"traits:{signature}")
    return {
        "alignment": rng.uniform(0.0, 1.0),
        "power": rng.uniform(0.0, 1.0),
        "volatility": rng.uniform(0.0, 1.0),
        "deceptiveness": rng.uniform(0.0, 1.0),
    }


def normalize_profile(profile: dict, signature: str) -> dict:
    """Coerce an LLM (or fallback) profile into the exact shape the DB and
    frontend expect, filling gaps with signature-deterministic defaults."""
    rng = random.Random(f"norm:{signature}")
    voice = profile.get("voice") if isinstance(profile.get("voice"), dict) else {}
    visual = profile.get("visual") if isinstance(profile.get("visual"), dict) else {}
    quotes = profile.get("quotes") if isinstance(profile.get("quotes"), list) else []

    rarity = str(profile.get("rarity", "common")).lower()
    if rarity not in RARITY_TIERS:
        rarity = "common"

    form = str(visual.get("form", "wisp")).lower()
    if form not in ("wisp", "banshee", "fairy", "shade"):
        form = "wisp"

    def _clamp(value, low, high, default):
        try:
            return max(low, min(high, float(value)))
        except (TypeError, ValueError):
            return default

    return {
        "name": str(profile.get("name", "The Unnamed"))[:64].strip() or "The Unnamed",
        "epithet": str(profile.get("epithet", rng.choice(_EPITHETS)))[:128],
        "persona": str(profile.get("persona", ""))[:1200],
        "rarity": rarity,
        "voice": {
            "voice_id": voice.get("voice_id")
            if voice.get("voice_id") in EN_VOICE_IDS + ["davefx", "ald"]
            else rng.choice(EN_VOICE_IDS),
            "pitch": _clamp(voice.get("pitch"), -6, 6, rng.uniform(-3, 3)),
            "rate": _clamp(voice.get("rate"), 0.8, 1.15, rng.uniform(0.9, 1.05)),
            "noise": _clamp(voice.get("noise"), 0.01, 0.08, rng.uniform(0.02, 0.05)),
            "echo": _clamp(voice.get("echo"), 0.0, 0.5, rng.uniform(0.1, 0.3)),
        },
        "visual": {
            "hue": _clamp(visual.get("hue"), 0, 360, rng.uniform(0, 360)),
            "form": form,
        },
        "quotes": [str(q)[:200] for q in quotes[:4] if isinstance(q, str)]
        or rng.sample(_QUOTE_BANK, 2),
        "traits": roll_traits(signature),
    }


def fallback_profile(signature: str) -> dict:
    """Procedural persona for when the LLM box is unreachable — a summoning
    must never visibly fail."""
    rng = random.Random(f"fallback:{signature}")
    name = rng.choice(_NAME_PARTS[0]) + rng.choice(_NAME_PARTS[1])
    epithet = rng.choice(_EPITHETS)
    persona = rng.choice(_PERSONA_TEMPLATES).format(name=name)
    rarity = rng.choices(RARITY_TIERS, weights=[55, 30, 12, 3])[0]
    return normalize_profile(
        {
            "name": name,
            "epithet": epithet,
            "persona": persona,
            "rarity": rarity,
            "quotes": rng.sample(_QUOTE_BANK, 2),
        },
        signature,
    )