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.entropy import veil_random
from app.models.entity import RARITY_TIERS
from app.tts.voices import EN_VOICE_IDS, ES_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


# --- voice archetypes ------------------------------------------------------
#
# Rolling pitch/rate/noise/echo independently produces random *effects*, not
# distinct *people*: you get a uniform smear of slightly-odd voices that all
# sound like the same synthesizer wearing different hats. Real voices covary
# — an old man is low AND slow AND breathy; a child is high AND quick AND
# clean. So a throat is drawn as one coherent bundle, and only then jittered
# slightly, which is what makes two spirits sound like two different dead
# people rather than two settings.
#
# `voices` lists which Piper models suit the archetype (see tts/voices.py);
# ranges are (low, high) sampled uniformly.
VOICE_ARCHETYPES: tuple[dict, ...] = (
    {
        "key": "elder_man",
        "voices": ("ryan", "alan", "hfc_male"),
        # Deep, slow, and worn: age thins the voice and adds breath.
        "pitch": (-8.5, -4.5),
        "rate": (0.80, 0.90),
        "noise": (0.045, 0.075),
        "echo": (0.15, 0.30),
    },
    {
        "key": "elder_woman",
        "voices": ("lessac", "hfc_female"),
        "pitch": (-3.0, 0.5),
        "rate": (0.82, 0.92),
        "noise": (0.040, 0.070),
        "echo": (0.15, 0.30),
    },
    {
        "key": "young_man",
        "voices": ("ryan", "hfc_male"),
        "pitch": (-2.0, 1.5),
        "rate": (1.00, 1.12),
        "noise": (0.015, 0.035),
        "echo": (0.05, 0.18),
    },
    {
        "key": "young_woman",
        "voices": ("amy", "lessac"),
        "pitch": (0.5, 4.0),
        "rate": (1.00, 1.14),
        "noise": (0.015, 0.035),
        "echo": (0.05, 0.18),
    },
    {
        "key": "child",
        # Highest pitch and quickest delivery; a child's voice is also the
        # cleanest, which is exactly what makes it unsettling through static.
        "voices": ("amy", "lessac"),
        "pitch": (6.0, 9.0),
        "rate": (1.08, 1.20),
        "noise": (0.010, 0.025),
        "echo": (0.20, 0.38),
    },
    {
        "key": "drowned",
        # Waterlogged: pitched down, dragging, and very wet with echo.
        "voices": ("hfc_female", "amy", "hfc_male"),
        "pitch": (-6.0, -2.0),
        "rate": (0.78, 0.88),
        "noise": (0.055, 0.085),
        "echo": (0.34, 0.50),
    },
    {
        "key": "burned",
        # Smoke-ruined throat: the noisiest archetype, barely holding pitch.
        "voices": ("hfc_male", "hfc_female", "alan"),
        "pitch": (-5.0, -1.0),
        "rate": (0.84, 0.96),
        "noise": (0.070, 0.100),
        "echo": (0.10, 0.24),
    },
    {
        "key": "distant",
        # Far away rather than damaged — quiet, even, drenched in space.
        "voices": ("lessac", "alan", "amy"),
        "pitch": (-2.0, 2.0),
        "rate": (0.88, 1.00),
        "noise": (0.025, 0.045),
        "echo": (0.40, 0.50),
    },
)


def roll_voice(rng: random.Random, language: str = "en") -> dict:
    """Draw one coherent throat.

    Spanish has only two installed models, so the archetype's voice list is
    ignored there and the shaping alone carries the difference — better a
    correctly-pronounced spirit with archetype shaping than an English
    model mangling Spanish text.
    """
    arch = rng.choice(VOICE_ARCHETYPES)
    if language == "es":
        voice_id = rng.choice(ES_VOICE_IDS)
    else:
        # Guard against an archetype naming a voice that isn't installed.
        candidates = [v for v in arch["voices"] if v in EN_VOICE_IDS] or EN_VOICE_IDS
        voice_id = rng.choice(candidates)
    return {
        "voice_id": voice_id,
        "pitch": rng.uniform(*arch["pitch"]),
        "rate": rng.uniform(*arch["rate"]),
        "noise": rng.uniform(*arch["noise"]),
        "echo": rng.uniform(*arch["echo"]),
        "archetype": arch["key"],
    }


def roll_traits(signature: str, entropy: object = None) -> dict:
    """Roll the four hidden truth-traits for an entity. 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.

    With `entropy` (a physical-noise contribution from the seeker's room —
    see app/entropy.py), the roll is genuinely unpredictable: what answers
    is decided by the room, not by a hash. Without it, the roll stays
    signature-deterministic, which is what the pure-function tests and
    `judgment`'s favor-bias comparisons rely on.

    A separate `random.Random` namespace (`"traits:"` vs. `normalize_profile`'s
    `"norm:"`) keeps this roll independent of the cosmetic-defaults rng."""
    if entropy is not None:
        # Domain-separated from the "which entity answers" draw so the two
        # can't be correlated — learning an entity's rarity must reveal
        # nothing about its hidden alignment.
        rng = veil_random(entropy, f"traits:{signature}")
    else:
        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, entropy: object = None) -> dict:
    """Coerce an LLM (or fallback) profile into the exact shape the DB and
    frontend expect, filling gaps with defaults.

    With `entropy`, the gap-filling defaults (voice, hue, form) are drawn
    from physical noise, so two spirits minted on the same channel don't
    inherit identical throats and colours. Without it, defaults stay
    signature-deterministic for the pure-function tests."""
    rng = (
        veil_random(entropy, f"norm:{signature}")
        if entropy is not None
        else 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 []

    throat = roll_voice(rng, "es" if profile.get("_language") == "es" else "en")

    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,
        # A coherent throat, not four independent knobs — see roll_voice().
        # The LLM may pin a voice_id and nudge parameters, but anything it
        # leaves out (or gets wrong) falls back to the archetype's own
        # covarying values rather than to a bland midpoint, so every spirit
        # lands somewhere recognisably human.
        "voice": {
            "voice_id": voice.get("voice_id")
            if voice.get("voice_id") in EN_VOICE_IDS + ES_VOICE_IDS
            else throat["voice_id"],
            "pitch": _clamp(voice.get("pitch"), -9, 9, throat["pitch"]),
            "rate": _clamp(voice.get("rate"), 0.75, 1.20, throat["rate"]),
            "noise": _clamp(voice.get("noise"), 0.01, 0.10, throat["noise"]),
            "echo": _clamp(voice.get("echo"), 0.0, 0.5, throat["echo"]),
            "archetype": throat["archetype"],
        },
        "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, entropy),
    }


# Base rarity weights (common, uncommon, rare, mythic). The moon skews
# these: a dark sky keeps the ordinary dead ordinary, a full moon drags the
# distribution toward the strange. Real illumination drives it — see
# app/celestial.py, which computes actual phase rather than approximating a
# calendar.
BASE_RARITY_WEIGHTS = (55, 30, 12, 3)


def rarity_weights_for_moon(illumination: float | None) -> tuple[float, ...]:
    """Skew the rarity table by how much of the moon is lit.

    At new moon the table is essentially the base one. At full moon the two
    rare tiers roughly triple while common falls away, so a mythic
    summoning stops being a lottery win and becomes a reason to go out on
    the right night — which is the entire point of tracking the phase.

    Deliberately a *skew*, never a gate: every tier stays reachable on
    every night, because a seeker who can only play on a Tuesday should
    never be locked out of the good spirits.
    """
    if illumination is None:
        return BASE_RARITY_WEIGHTS
    lit = max(0.0, min(1.0, illumination))
    common, uncommon, rare, mythic = BASE_RARITY_WEIGHTS
    return (
        common * (1.0 - 0.55 * lit),
        uncommon * (1.0 + 0.10 * lit),
        rare * (1.0 + 1.60 * lit),
        mythic * (1.0 + 2.20 * lit),
    )


def moon_trait_bias(traits: dict, illumination: float | None) -> dict:
    """Nudge hidden traits by moonlight.

    Folklore is consistent that a full moon brings out what is strongest
    and least stable, so power and volatility rise with illumination while
    alignment drifts very slightly darker. Kept small (<=0.12) and clamped:
    this is a thumb on the scale, not a rewrite — a benevolent spirit under
    a full moon is still benevolent, just more so of whatever it already
    was.
    """
    if illumination is None:
        return traits
    lit = max(0.0, min(1.0, illumination))

    def _clamp01(v: float) -> float:
        return max(0.0, min(1.0, v))

    return {
        **traits,
        "power": _clamp01(traits.get("power", 0.5) + 0.12 * lit),
        "volatility": _clamp01(traits.get("volatility", 0.5) + 0.10 * lit),
        "alignment": _clamp01(traits.get("alignment", 0.5) - 0.06 * lit),
    }


def fallback_profile(
    signature: str, entropy: object = None, illumination: float | None = None
) -> dict:
    """Procedural persona for when the LLM box is unreachable — a summoning
    must never visibly fail.

    With `entropy`, even the fallback spirits differ run to run: with the
    LLM down, a purely signature-seeded fallback would hand every seeker on
    a given channel the identical name, epithet and rarity, which is
    exactly the "example data" feel this is meant to avoid."""
    rng = (
        veil_random(entropy, f"fallback:{signature}")
        if entropy is not None
        else 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=rarity_weights_for_moon(illumination))[0]
    return normalize_profile(
        {
            "name": name,
            "epithet": epithet,
            "persona": persona,
            "rarity": rarity,
            "quotes": rng.sample(_QUOTE_BANK, 2),
        },
        signature,
        entropy,
    )