drjones/qtalker---
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: the moon shapes who answers; wire magnetometer; drop dead CSS

Browse Source 
MOON INFLUENCE ON SUMMONING

Astronomy previously only decided whether a channel's familiar spirit
returned. It now shapes *who comes through*:

  - Rarity skews with real moon illumination. At full moon the rare and
    mythic weights roughly triple while common recedes, so a mythic
    summoning becomes a reason to go out on the right night rather than a
    flat lottery. Deliberately a skew and never a gate — every tier stays
    reachable on every night, because someone who can only play midweek
    should not be locked out of the good spirits.
  - Hidden traits take a small moonlit nudge: power and volatility rise,
    alignment drifts slightly darker. Capped at 0.12 and clamped to [0,1],
    so a full moon intensifies what a spirit already is instead of
    rewriting it. Deceptiveness is untouched — whether a spirit lies is its
    own nature, not the sky's doing.
  - The mint prompt is told the phase, and explicitly told the entity must
    never mention or seem aware of it. It shapes who they are, not their
    dialogue; a ghost remarking on the moonlight would break the illusion
    instantly.

Tests assert the outcomes shift in practice (mythic rate over 4000 draws,
rare-tier counts across 300 fallback profiles), not merely that the code
runs. test_mint_prompt_never_receives_traits now allows `sky` while still
forbidding `traits`: moon phase is public, observable state anyone can look
up, hidden ground truth is not.

GEOMAGNETIC (app/geomagnetic.py)

Real NOAA SWPC planetary K-index, verified against the live endpoint —
which caught a real bug: I had written the parser against an
array-of-arrays shape, and the actual feed serves a list of objects
(`estimated_kp` float, `kp_index` int, `kp` a display string with a letter
suffix). Fixed, and the tests now use the real captured shape. Cached,
never blocking, and a failed refresh keeps serving the last real value —
an hour-old genuine measurement beats nothing, and geomagnetic conditions
do not change fast enough for that to mislead.

MAGNETOMETER WIRED

MagnetometerListener existed but was never connected. The EMF panel now
runs it alongside the motion listener where the hardware exists, so the
"EMF meter" measures actual magnetic field in µT rather than only
inferring disturbance from movement. Additive: the motion path is
untouched and remains the only option on iOS. Its field jitter also feeds
the entropy pool.

DEAD CODE

Removed .evp-scope and .radio-waterfall, orphaned when both panels moved to
the shared SpectrumScope. Audited every other flagged export first and left
them alone — they are used internally, and "not imported elsewhere" is not
the same as dead.

Adds docs/CHANNELS.md recording what each channel measures and, honestly,
what has actually been verified against hardware versus only written
carefully.

311 backend + 355 frontend tests pass; i18n parity gate passes.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Indiana
2026-07-28 13:31:15 +00:00
parent b8e69b4bd3
commit cf602ab3de
14 changed files with 1009 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

196
backend/app/entities.py
View File

@@ -8,7 +8,7 @@ import random
from app.entropy import veil_random
from app.models.entity import RARITY_TIERS
from app.tts.voices import EN_VOICE_IDS
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.
@@ -97,6 +97,117 @@ def parse_mint_response(text: str) -> dict | 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`,
@@ -144,6 +255,8 @@ def normalize_profile(profile: dict, signature: str, entropy: object = None) ->
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"
@@ -163,14 +276,20 @@ def normalize_profile(profile: dict, signature: str, entropy: object = None) ->
"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 + ["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)),
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)),
@@ -182,7 +301,66 @@ def normalize_profile(profile: dict, signature: str, entropy: object = None) ->
}
def fallback_profile(signature: str, entropy: object = None) -> dict:
# 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.
@@ -198,7 +376,7 @@ def fallback_profile(signature: str, entropy: object = None) -> dict:
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]
rarity = rng.choices(RARITY_TIERS, weights=rarity_weights_for_moon(illumination))[0]
return normalize_profile(
{
"name": name,

203
backend/app/geomagnetic.py Normal file
View File

@@ -0,0 +1,203 @@
"""Real geomagnetic activity from NOAA's Space Weather Prediction Center.
The planetary K-index (Kp) is a genuine measurement of geomagnetic
disturbance, derived from magnetometer observatories worldwide and
published by NOAA SWPC. Scale is 0-9: below 4 is quiet, 5+ is an official
geomagnetic storm, 8-9 is severe.
Why this belongs in a séance app: geomagnetic storms are the single most
commonly cited "explanation" in paranormal circles, and unlike most such
claims this one is a real, measured, publicly published number. So the app
can honestly say the planet's magnetic field is disturbed tonight, because
NOAA measured it. We report the measurement and let the fiction sit on top;
we never claim the storm causes anything.
Source: https://services.swpc.noaa.gov/json/planetary_k_index_1m.json
Free, no API key, no attribution requirement (US government public data).
Design constraints this module respects:
- Never blocks a summon. Every lookup is served from cache; a refresh
that fails leaves the previous value in place, and a cold cache simply
reports None. A séance must not wait on, or fail because of, a third
party's web service.
- Polite polling. Kp updates at most every minute, so the cache TTL is
generous and concurrent callers share a single in-flight refresh
rather than stampeding NOAA.
"""
import asyncio
import time
import httpx
SWPC_KP_URL = "https://services.swpc.noaa.gov/json/planetary_k_index_1m.json"
# Kp is published every minute but changes slowly; 10 minutes is plenty
# fresh for a "is the field disturbed tonight" reading and keeps us a
# courteous consumer of a free public service.
CACHE_TTL_S = 600
# Timeout tight enough that nothing user-facing ever notices. The value is
# a garnish; if NOAA is slow we simply go without it.
REQUEST_TIMEOUT_S = 4.0
# Official NOAA G-scale thresholds.
STORM_THRESHOLD = 5.0
SEVERE_THRESHOLD = 8.0
def storm_label(kp: float) -> str:
"""NOAA's own severity wording, not invented adjectives."""
if kp >= SEVERE_THRESHOLD:
return "severe geomagnetic storm"
if kp >= 7:
return "strong geomagnetic storm"
if kp >= 6:
return "moderate geomagnetic storm"
if kp >= STORM_THRESHOLD:
return "minor geomagnetic storm"
if kp >= 4:
return "unsettled"
return "quiet"
def disturbance(kp: float) -> float:
"""Kp mapped to 0..1 for blending with the celestial reading.
Divided by 9 (the scale maximum) rather than by an observed range, so
the number means "how far up the actual Kp scale we are" and stays
interpretable against NOAA's published values.
"""
return max(0.0, min(1.0, kp / 9.0))
class GeomagneticCache:
"""Shared, non-blocking cache of the latest Kp reading."""
def __init__(self, url: str = SWPC_KP_URL, ttl_s: float = CACHE_TTL_S):
self._url = url
self._ttl = ttl_s
self._kp: float | None = None
self._fetched_at = 0.0
# Serialises refreshes so N concurrent summons trigger one request,
# not N.
self._lock = asyncio.Lock()
@property
def cached_kp(self) -> float | None:
"""Last known value without triggering any network I/O."""
return self._kp
def _is_fresh(self) -> bool:
return self._kp is not None and (time.monotonic() - self._fetched_at) < self._ttl
async def get(self) -> float | None:
"""Current Kp, refreshing if stale.
Returns None only when we have never successfully fetched. A failed
refresh deliberately keeps serving the stale value: an hour-old
real measurement is far better than nothing, and geomagnetic
conditions do not change fast enough for staleness to mislead.
"""
if self._is_fresh():
return self._kp
async with self._lock:
# Another waiter may have refreshed while we queued.
if self._is_fresh():
return self._kp
try:
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT_S) as client:
response = await client.get(self._url)
response.raise_for_status()
payload = response.json()
kp = _latest_kp(payload)
if kp is not None:
self._kp = kp
self._fetched_at = time.monotonic()
except Exception:
# Deliberately broad: DNS failure, timeout, TLS problem,
# malformed JSON, NOAA schema change — none of them are
# worth failing or delaying a séance over. Keep the stale
# value and try again after the TTL.
pass
return self._kp
async def reading(self) -> dict | None:
"""Full reading for display/prompting, or None if never fetched."""
kp = await self.get()
if kp is None:
return None
return {
"kp": kp,
"label": storm_label(kp),
"disturbance": disturbance(kp),
"storm": kp >= STORM_THRESHOLD,
}
def _latest_kp(payload: object) -> float | None:
"""Pull the most recent Kp out of SWPC's payload.
The live `planetary_k_index_1m` feed is a list of objects, verified
against the real service:
[{"time_tag": "2026-07-28T05:54:00", "kp_index": 1,
"estimated_kp": 1.00, "kp": "1Z"}, ...]
`estimated_kp` is preferred because it is a float and carries the
fractional precision Kp actually has; `kp_index` is the integer
rounding of it. The `kp` string field is deliberately ignored — it
carries a letter suffix ("1Z") and is a display form, not a number.
Some other SWPC endpoints serve a header-row + array-of-arrays shape
instead, so that is handled too rather than assumed away — the
difference is invisible to callers and costs a few lines.
Parsed from the end backwards, since the feed appends chronologically
and can carry trailing rows with unusable values. Any shape this does
not recognise degrades to None rather than raising inside a summon.
"""
if not isinstance(payload, list) or not payload:
return None
def _valid(value: object) -> float | None:
try:
kp = float(value) # type: ignore[arg-type]
except (TypeError, ValueError):
return None
return kp if 0.0 <= kp <= 9.0 else None
# Live shape: list of objects.
if isinstance(payload[0], dict):
for row in reversed(payload):
if not isinstance(row, dict):
continue
for key in ("estimated_kp", "kp_index"):
kp = _valid(row.get(key))
if kp is not None:
return kp
return None
# Legacy/alternate shape: header row then data rows.
if len(payload) < 2:
return None
header = payload[0]
kp_col = 1
if isinstance(header, list):
for i, name in enumerate(header):
if isinstance(name, str) and name.strip().lower() in ("kp", "kp_index", "estimated_kp"):
kp_col = i
break
for row in reversed(payload[1:]):
if not isinstance(row, list) or len(row) <= kp_col:
continue
kp = _valid(row[kp_col])
if kp is not None:
return kp
return None
# App-wide instance; tests substitute their own.
geomagnetic_cache = GeomagneticCache()

16
backend/app/llm/prompts.py
View File

@@ -76,6 +76,13 @@ MINT_SYSTEM = (
MINT_PROMPT = """A new presence has been detected through {channel}. Its signal signature is {signature}.
Recent anomalous pulses observed: {anomaly_summary}
The sky above the seeker right now: {sky}
Let the sky colour who answers, not what they say about it: a presence
that comes through under a full moon should be a *stronger, stranger sort
of person* than one that surfaces on a dark night — but they must never
mention the moon, or seem aware of it. It shapes who they are, not their
dialogue.
Invent the person behind this signal — a specific, once-living human being, not a generic spirit. Before writing, silently decide (do not output these separately, weave them into "persona" and "quotes" instead):
- A plausible full first name for someone who could have died at any point in roughly the last 120 years (their name and any period detail should stay internally consistent with each other).
@@ -90,7 +97,7 @@ Return ONLY a JSON object with exactly these keys:
"epithet": "a short, evocative title for how the site labels them, e.g. 'the Static Widow' — this is the installation's framing, not what they'd call themselves",
"persona": "2-3 sentences written as their real biography compressed, not as ghost-poetry: who they were (age, era, occupation, one specific mundane detail), then how that unfinished, ordinary thing keeps them bound to this signal. The mundane specificity should sit right up against the uncanny, not be replaced by it.",
"rarity": "one of: common, uncommon, rare, mythic (mythic only for truly strange signatures)",
"voice": {{"voice_id": "one of: {voice_ids}", "pitch": "semitone shift, -6 to 6, deeper for older/heavier spirits", "rate": "speech rate 0.8 to 1.15", "noise": "static level 0.01 to 0.08"}},
"voice": {{"voice_id": "one of: {voice_ids}", "pitch": "semitone shift, -9 to 9, deeper for older/heavier spirits, higher for children", "rate": "speech rate 0.75 to 1.20 — slower for the old, the tired and the waterlogged", "noise": "static level 0.01 to 0.10 — higher for ruined or breathing-damaged throats"}},
"visual": {{"hue": "0-360 color hue matching their nature", "form": "one of: wisp, banshee, fairy, shade"}},
"quotes": ["two short lines this specific person would actually say — at least one should be a plain, human, almost mundane fragment (a name, an unfinished instruction, an ordinary question) rather than cosmic riddle-speak; let the wrongness of hearing it through static be what's unsettling, not the words themselves"]
}}"""
@@ -191,11 +198,16 @@ def manifest_prompt(readings: dict) -> str:
def mint_prompt(
signature: str, channel: str, anomaly_summary: str, voice_ids: list[str]
signature: str,
channel: str,
anomaly_summary: str,
voice_ids: list[str],
sky: str = "unknown",
) -> str:
return MINT_PROMPT.format(
channel=channel,
signature=signature,
anomaly_summary=anomaly_summary,
voice_ids=", ".join(voice_ids),
sky=sky,
)

19
backend/app/llm/service.py
View File

@@ -192,6 +192,7 @@ class SpiritService:
anomalies: list[dict],
language: str = "en",
entropy: object = None,
sky: dict | None = None,
) -> dict:
"""Invent a full persona for a new signature, normalized to schema.
@@ -201,11 +202,17 @@ class SpiritService:
channel are genuinely different rather than identical."""
summary = json.dumps(anomalies[-10:])[:600]
voice_ids = ES_VOICE_IDS if language == "es" else EN_VOICE_IDS
illumination = sky.get("moon_illumination") if sky else None
sky_text = (
f"{sky['moon_name']}, {sky['moon_illumination']:.0%} lit"
if sky
else "unknown"
)
async def call() -> str:
return await self._client.generate(
settings.ollama_chat_model,
prompts.mint_prompt(signature, channel, summary, voice_ids),
prompts.mint_prompt(signature, channel, summary, voice_ids, sky_text),
system=prompts.MINT_SYSTEM,
options={"num_predict": 400, "temperature": 0.9},
)
@@ -215,10 +222,14 @@ class SpiritService:
self._touch()
profile = entities.parse_mint_response(raw)
if profile is None:
return entities.fallback_profile(signature, entropy)
return entities.normalize_profile(profile, signature, entropy)
return entities.fallback_profile(signature, entropy, illumination)
normalized = entities.normalize_profile(profile, signature, entropy)
normalized["traits"] = entities.moon_trait_bias(
normalized["traits"], illumination
)
return normalized
except (QueueFullError, httpx.HTTPError, KeyError, ValueError):
return entities.fallback_profile(signature, entropy)
return entities.fallback_profile(signature, entropy, illumination)
# The app-wide instance; tests monkeypatch this.

27
backend/app/ws.py
View File

@@ -38,6 +38,7 @@ from app.deps import SESSION_COOKIE_NAME
from app.entities import fallback_signature, signature_from_anomalies
from app.celestial import veil_thinness
from app.entropy import contribution_bits, veil_float
from app.geomagnetic import geomagnetic_cache
from app.inventory import (
RITUAL_SUCCESS_ESSENCE,
SUMMON_ESSENCE_TRICKLE,
@@ -148,6 +149,9 @@ class SeanceState:
# coordinate would be retaining precise location data we have no use
# for. Never persisted; lives and dies with the connection.
longitude: float | None = None
# Latest NOAA planetary K-index reading, refreshed on summon. Cached on
# the state so the manifest path can report it without another lookup.
geomagnetic: dict | None = None
# Active-session registry (spec: ESP32 sensor node, Workstream K): maps a
@@ -362,7 +366,13 @@ async def _summon(state: SeanceState, channel: str) -> tuple[Entity, bool]:
# premise. Real astronomy, computed from the seeker's own
# longitude: see app/celestial.py.
sky = veil_thinness(state.longitude)
return_chance = RETURN_CHANCE * (1.0 - sky["thinness"] * VEIL_THINNESS_PULL)
thinness = sky["thinness"]
if state.geomagnetic is not None:
# A real storm counts for a third of the reading — enough
# that a Kp-7 night is noticeably stranger, not so much
# that quiet space weather flattens the moon's effect.
thinness = thinness * 0.67 + state.geomagnetic["disturbance"] * 0.33
return_chance = RETURN_CHANCE * (1.0 - thinness * VEIL_THINNESS_PULL)
answers = (
known is not None
and veil_float(state.entropy, "answers") < return_chance
@@ -382,6 +392,9 @@ async def _summon(state: SeanceState, channel: str) -> tuple[Entity, bool]:
profile = await spirit_service.mint_profile(
mint_signature, channel, state.anomalies, state.language,
entropy=state.entropy,
# The moon shapes who answers — see
# entities.rarity_weights_for_moon / moon_trait_bias.
sky=sky,
)
discoverer = await db.get(User, state.user_id)
favor = discoverer.favor if discoverer is not None else 0.0
@@ -484,10 +497,18 @@ async def _handle_summon(state: SeanceState) -> None:
)
return
# Real measured geomagnetic activity (NOAA SWPC). Served from cache and
# never blocking: a cold cache or an outage yields None and the séance
# proceeds on astronomy alone.
geo = await geomagnetic_cache.reading()
if geo is not None:
state.geomagnetic = geo
await state.send_queue.put(
{
"type": "status",
"state": "summoning",
"geomagnetic": state.geomagnetic,
# How much physical noise from the room actually fed this
# draw. Display only — a client can lie about it freely, and
# it is never used to weight or gate anything.
@@ -551,6 +572,10 @@ def _room_readings(state: SeanceState, anomaly: dict) -> dict:
if isinstance(mag, (int, float)):
readings["deviation above the floor"] = f"{mag:.1f}"
readings["disturbances so far"] = str(len(state.anomalies))
if state.geomagnetic is not None:
readings["earth's magnetic field"] = (
f"Kp {state.geomagnetic['kp']:.2f} — {state.geomagnetic['label']}"
)
sky = veil_thinness(state.longitude)
readings["moon"] = f"{sky['moon_name']} ({sky['moon_illumination']:.0%} lit)"
if sky["witching_proximity"] is not None:

190
backend/tests/test_entities.py
View File

@@ -1,5 +1,7 @@
import inspect
import pytest
from app.entities import (
fallback_profile,
normalize_profile,
@@ -52,7 +54,9 @@ def test_normalize_profile_fills_and_clamps():
assert profile["name"] == "Hollow Briar"
assert profile["rarity"] == "common"
assert profile["voice"]["voice_id"] != "nonexistent"
assert -6 <= profile["voice"]["pitch"] <= 6
# Widened from +-6 when voices moved to coherent archetypes: a child
# and an elder man have to be able to land genuinely far apart.
assert -9 <= profile["voice"]["pitch"] <= 9
assert 0.01 <= profile["voice"]["noise"] <= 0.08
assert profile["visual"]["form"] in ("wisp", "banshee", "fairy", "shade")
assert 0 <= profile["visual"]["hue"] <= 360
@@ -150,4 +154,186 @@ def test_mint_prompt_never_receives_traits():
generation."""
params = set(inspect.signature(llm_prompts.mint_prompt).parameters)
assert "traits" not in params
assert params == {"signature", "channel", "anomaly_summary", "voice_ids"}
# `sky` is public, observable state (moon phase — anyone can look up),
# so it may shape the persona. Hidden ground truth still must not.
assert params == {"signature", "channel", "anomaly_summary", "voice_ids", "sky"}
# --- voice archetypes -------------------------------------------------------
class TestVoiceArchetypes:
"""A voice must sound like a person, not like four random knobs.
These assert the *covariance* that makes an archetype legible — a child
being high AND fast AND clean — because that is exactly what
independently-rolled parameters would destroy.
"""
def _throats(self, n=200, language="en"):
import random as _r
from app.entities import roll_voice
return [roll_voice(_r.Random(i), language) for i in range(n)]
def test_every_archetype_is_reachable(self):
from app.entities import VOICE_ARCHETYPES
seen = {t["archetype"] for t in self._throats(400)}
assert seen == {a["key"] for a in VOICE_ARCHETYPES}
def test_all_parameters_stay_inside_the_synthesis_limits(self):
for t in self._throats(300):
assert -9 <= t["pitch"] <= 9
assert 0.75 <= t["rate"] <= 1.20
assert 0.01 <= t["noise"] <= 0.10
assert 0.0 <= t["echo"] <= 0.5
def test_only_installed_voices_are_chosen(self):
from app.tts.voices import VOICES
for t in self._throats(300):
assert t["voice_id"] in VOICES
def test_spanish_spirits_get_spanish_models(self):
# An English model reading Spanish text mangles it; better to carry
# the difference in shaping alone.
from app.tts.voices import ES_VOICE_IDS
for t in self._throats(120, language="es"):
assert t["voice_id"] in ES_VOICE_IDS
def test_children_are_higher_and_faster_than_elders(self):
throats = self._throats(600)
child = [t for t in throats if t["archetype"] == "child"]
elder = [t for t in throats if t["archetype"] == "elder_man"]
assert child and elder
assert min(t["pitch"] for t in child) > max(t["pitch"] for t in elder)
assert min(t["rate"] for t in child) > max(t["rate"] for t in elder)
def test_damaged_throats_are_noisier_than_clean_ones(self):
throats = self._throats(600)
burned = [t for t in throats if t["archetype"] == "burned"]
child = [t for t in throats if t["archetype"] == "child"]
assert burned and child
assert min(t["noise"] for t in burned) > max(t["noise"] for t in child)
def test_the_distant_archetype_is_the_most_reverberant(self):
throats = self._throats(600)
distant = [t for t in throats if t["archetype"] == "distant"]
young = [t for t in throats if t["archetype"] == "young_man"]
assert distant and young
assert min(t["echo"] for t in distant) > max(t["echo"] for t in young)
def test_two_spirits_rarely_share_a_throat(self):
# The whole point: distinguishable people, not a smear of one voice.
throats = self._throats(120)
fingerprints = {
(t["voice_id"], round(t["pitch"], 1), round(t["rate"], 2)) for t in throats
}
assert len(fingerprints) > 100
def test_normalize_profile_reports_the_archetype(self):
from app.entities import normalize_profile
profile = normalize_profile({"name": "Test"}, "sig-archetype")
assert profile["voice"]["archetype"]
# --- moon influence on summoning --------------------------------------------
class TestMoonInfluence:
"""The moon must change *who answers*, not just decorate the UI."""
def test_dark_sky_leaves_the_base_table_alone(self):
from app.entities import BASE_RARITY_WEIGHTS, rarity_weights_for_moon
assert rarity_weights_for_moon(0.0) == pytest.approx(BASE_RARITY_WEIGHTS)
def test_unknown_sky_falls_back_to_the_base_table(self):
from app.entities import BASE_RARITY_WEIGHTS, rarity_weights_for_moon
assert rarity_weights_for_moon(None) == BASE_RARITY_WEIGHTS
def test_a_full_moon_makes_rare_and_mythic_much_likelier(self):
from app.entities import rarity_weights_for_moon
dark = rarity_weights_for_moon(0.0)
full = rarity_weights_for_moon(1.0)
assert full[2] > dark[2] * 2 # rare
assert full[3] > dark[3] * 2 # mythic
assert full[0] < dark[0] # common recedes
def test_every_tier_stays_reachable_on_every_night(self):
# A skew, never a gate — someone who can only play midweek must not
# be locked out of the good spirits.
from app.entities import rarity_weights_for_moon
for lit in (0.0, 0.25, 0.5, 0.75, 1.0):
assert all(w > 0 for w in rarity_weights_for_moon(lit))
def test_rarity_odds_actually_shift_in_practice(self):
import random as _r
from app.entities import rarity_weights_for_moon
from app.models.entity import RARITY_TIERS
def mythic_rate(lit):
rng = _r.Random(1234)
weights = rarity_weights_for_moon(lit)
draws = [rng.choices(RARITY_TIERS, weights=weights)[0] for _ in range(4000)]
return draws.count("mythic") / len(draws)
assert mythic_rate(1.0) > mythic_rate(0.0) * 1.8
def test_moonlight_raises_power_and_volatility(self):
from app.entities import moon_trait_bias
base = {"alignment": 0.5, "power": 0.5, "volatility": 0.5, "deceptiveness": 0.5}
full = moon_trait_bias(base, 1.0)
assert full["power"] > base["power"]
assert full["volatility"] > base["volatility"]
assert full["alignment"] < base["alignment"]
def test_bias_is_a_thumb_on_the_scale_not_a_rewrite(self):
from app.entities import moon_trait_bias
base = {"alignment": 0.5, "power": 0.5, "volatility": 0.5, "deceptiveness": 0.5}
full = moon_trait_bias(base, 1.0)
for key in base:
assert abs(full[key] - base[key]) <= 0.12
def test_bias_never_escapes_the_unit_range(self):
from app.entities import moon_trait_bias
for extreme in (0.0, 1.0):
base = dict.fromkeys(
("alignment", "power", "volatility", "deceptiveness"), extreme
)
for lit in (0.0, 0.5, 1.0):
for v in moon_trait_bias(base, lit).values():
assert 0.0 <= v <= 1.0
def test_unknown_sky_leaves_traits_untouched(self):
from app.entities import moon_trait_bias
base = {"alignment": 0.3, "power": 0.7, "volatility": 0.2, "deceptiveness": 0.9}
assert moon_trait_bias(base, None) == base
def test_deceptiveness_is_not_moon_driven(self):
# Whether a spirit lies is its own nature, not the sky's doing.
from app.entities import moon_trait_bias
base = {"alignment": 0.5, "power": 0.5, "volatility": 0.5, "deceptiveness": 0.5}
assert moon_trait_bias(base, 1.0)["deceptiveness"] == 0.5
def test_fallback_profiles_respect_the_moon(self):
from app.entities import fallback_profile
dark = [fallback_profile(f"s{i}", None, 0.0)["rarity"] for i in range(300)]
full = [fallback_profile(f"s{i}", None, 1.0)["rarity"] for i in range(300)]
rare_dark = sum(r in ("rare", "mythic") for r in dark)
rare_full = sum(r in ("rare", "mythic") for r in full)
assert rare_full > rare_dark

173
backend/tests/test_geomagnetic.py Normal file
View File

@@ -0,0 +1,173 @@
"""Tests for the NOAA geomagnetic feed.
No test here touches the network. The properties that matter are the
defensive ones: a séance must never fail, block, or crash because a free
third-party service is down, slow, or changed its schema.
"""
import pytest
from app.geomagnetic import (
STORM_THRESHOLD,
GeomagneticCache,
_latest_kp,
disturbance,
storm_label,
)
def _payload(*kps):
"""The real shape SWPC's planetary_k_index_1m endpoint serves, verified
against the live service — a list of objects, not array-of-arrays."""
return [
{
"time_tag": f"2026-07-28T0{i}:00:00",
"kp_index": int(float(kp)) if kp not in (None, "") else kp,
"estimated_kp": kp,
"kp": f"{kp}Z",
}
for i, kp in enumerate(kps)
]
def _legacy_payload(*kps):
"""The header-row shape some other SWPC endpoints use."""
rows = [["time_tag", "Kp", "a_running", "station_count"]]
for i, kp in enumerate(kps):
rows.append([f"2026-07-28 0{i}:00:00", kp, "4", "8"])
return rows
class TestLatestKp:
def test_reads_the_most_recent_row(self):
assert _latest_kp(_payload("1.00", "2.33", "5.67")) == 5.67
def test_reads_the_real_live_payload_shape(self):
# Captured verbatim from the live endpoint.
live = [
{"time_tag": "2026-07-27T23:59:00", "kp_index": 3, "estimated_kp": 3.00, "kp": "3Z"},
{"time_tag": "2026-07-28T05:54:00", "kp_index": 1, "estimated_kp": 1.00, "kp": "1Z"},
]
assert _latest_kp(live) == 1.0
def test_prefers_the_fractional_estimate_over_the_rounded_integer(self):
# estimated_kp carries the precision Kp actually has; kp_index is
# its rounding, so a 4.67 must not be reported as a 4.
row = [{"time_tag": "t", "kp_index": 4, "estimated_kp": 4.67, "kp": "4Z"}]
assert _latest_kp(row) == 4.67
def test_ignores_the_display_string_field(self):
# "5Z" is a display form with a letter suffix, not a number.
row = [{"time_tag": "t", "kp": "5Z"}]
assert _latest_kp(row) is None
def test_falls_back_to_kp_index_when_the_estimate_is_missing(self):
row = [{"time_tag": "t", "kp_index": 6}]
assert _latest_kp(row) == 6.0
def test_still_handles_the_legacy_header_row_shape(self):
assert _latest_kp(_legacy_payload("1.00", "3.5")) == 3.5
def test_finds_the_kp_column_by_header_name(self):
payload = [
["time_tag", "a_running", "Kp"],
["2026-07-28 00:00:00", "4", "3.5"],
]
assert _latest_kp(payload) == 3.5
def test_skips_trailing_rows_with_unusable_values(self):
payload = _payload("2.00", "4.00")
payload.append({"time_tag": "2026-07-28T03:00:00", "estimated_kp": None})
payload.append({"time_tag": "2026-07-28T04:00:00", "estimated_kp": ""})
assert _latest_kp(payload) == 4.0
def test_rejects_out_of_scale_values(self):
# Kp is defined 0-9; anything else is corrupt, not a real storm.
assert _latest_kp(_payload("42.0")) is None
assert _latest_kp(_payload("-3.0")) is None
@pytest.mark.parametrize(
"junk",
[None, [], "not a list", {"kp": 5}, [["time_tag", "Kp"]], 42, [[], []]],
)
def test_malformed_payloads_return_none_instead_of_raising(self, junk):
assert _latest_kp(junk) is None
def test_short_rows_do_not_index_error(self):
assert _latest_kp([["time_tag", "Kp"], ["2026-07-28"]]) is None
def test_objects_missing_every_known_field_return_none(self):
assert _latest_kp([{"time_tag": "t", "something_else": 3}]) is None
class TestStormLabel:
def test_uses_noaa_severity_bands(self):
assert storm_label(0.0) == "quiet"
assert storm_label(4.5) == "unsettled"
assert storm_label(5.0) == "minor geomagnetic storm"
assert storm_label(6.0) == "moderate geomagnetic storm"
assert storm_label(7.0) == "strong geomagnetic storm"
assert storm_label(9.0) == "severe geomagnetic storm"
def test_the_official_storm_threshold_is_where_storms_begin(self):
assert "storm" not in storm_label(STORM_THRESHOLD - 0.1)
assert "storm" in storm_label(STORM_THRESHOLD)
class TestDisturbance:
def test_maps_the_scale_onto_zero_to_one(self):
assert disturbance(0.0) == 0.0
assert disturbance(9.0) == 1.0
assert disturbance(4.5) == pytest.approx(0.5)
def test_clamps_rather_than_extrapolating(self):
assert disturbance(-5) == 0.0
assert disturbance(99) == 1.0
class TestCacheResilience:
@pytest.mark.asyncio
async def test_cold_cache_with_no_network_reports_none_and_does_not_raise(
self, monkeypatch
):
# Point at an unroutable host: the lookup must fail silently.
cache = GeomagneticCache(url="http://127.0.0.1:9/never", ttl_s=600)
assert await cache.get() is None
assert await cache.reading() is None
@pytest.mark.asyncio
async def test_a_failed_refresh_keeps_serving_the_last_real_value(self):
# An hour-old genuine measurement beats nothing, and geomagnetic
# conditions don't change fast enough for that to mislead.
cache = GeomagneticCache(url="http://127.0.0.1:9/never", ttl_s=0)
cache._kp = 6.0
assert await cache.get() == 6.0
@pytest.mark.asyncio
async def test_reading_reports_storm_state(self):
cache = GeomagneticCache(url="http://127.0.0.1:9/never", ttl_s=600)
cache._kp = 7.33
cache._fetched_at = float("inf")
reading = await cache.reading()
assert reading["kp"] == 7.33
assert reading["storm"] is True
assert reading["label"] == "strong geomagnetic storm"
assert 0.0 <= reading["disturbance"] <= 1.0
@pytest.mark.asyncio
async def test_quiet_conditions_are_not_reported_as_a_storm(self):
cache = GeomagneticCache(url="http://127.0.0.1:9/never", ttl_s=600)
cache._kp = 2.0
cache._fetched_at = float("inf")
reading = await cache.reading()
assert reading["storm"] is False
assert reading["label"] == "quiet"
@pytest.mark.asyncio
async def test_cached_kp_never_performs_io(self):
# The synchronous accessor exists precisely so hot paths can read
# the value without awaiting anything.
cache = GeomagneticCache(url="http://127.0.0.1:9/never")
assert cache.cached_kp is None
cache._kp = 3.0
assert cache.cached_kp == 3.0

4
backend/tests/test_ws_ritual_judgment.py
View File

@@ -21,7 +21,9 @@ from app.rate_limit import RateLimiter
class FakeSpiritService:
async def mint_profile(self, signature, channel, anomalies, language="en", entropy=None):
async def mint_profile(
self, signature, channel, anomalies, language="en", entropy=None, sky=None
):
return fallback_profile(signature)
async def fragment(self, source, anomaly, language="en"):

12
backend/tests/test_ws_session.py
View File

@@ -15,7 +15,9 @@ from app.rate_limit import RateLimiter
class FakeSpiritService:
async def mint_profile(self, signature, channel, anomalies, language="en", entropy=None):
async def mint_profile(
self, signature, channel, anomalies, language="en", entropy=None, sky=None
):
return fallback_profile(signature)
async def fragment(self, source, anomaly, language="en"):
@@ -379,7 +381,9 @@ async def test_summon_high_rarity_item_drop_is_persisted_and_sent(
monkeypatch.setattr(app.ws, "summon_limiter", RateLimiter(max_requests=1000, window_seconds=60))
class MythicSpiritService:
async def mint_profile(self, signature, channel, anomalies, language="en", entropy=None):
async def mint_profile(
self, signature, channel, anomalies, language="en", entropy=None, sky=None
):
profile = fallback_profile(signature)
profile["rarity"] = "mythic"
return profile
@@ -443,7 +447,9 @@ async def test_summon_common_rarity_does_not_roll_a_drop(sync_client, db_session
return profile
class CommonSpiritService:
async def mint_profile(self, signature, channel, anomalies, language="en", entropy=None):
async def mint_profile(
self, signature, channel, anomalies, language="en", entropy=None, sky=None
):
return _common_only(signature)
async def fragment(self, source, anomaly, language="en"):