drjones/qtalker---
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7d3a6e695dd96c53c8a481eb1e78232b5ffbc4c8
qtalker---/backend/tests/test_ws_ritual_judgment.py
Indiana cf602ab3de feat: the moon shapes who answers; wire magnetometer; drop dead CSS 
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>
2026-07-28 13:31:15 +00:00

621 lines
22 KiB
Python
Raw Blame History

"""Workstream B WS integration tests: ritual_start/ritual_step/judgment
handlers in app.ws, plus the favor/essence/at_peace side effects and the
favor read-back bias on trait rolls at mint time."""
import uuid
import pytest
from sqlalchemy import select
import app.judgment as judgment_module
import app.ws
from app.entities import fallback_profile
from app.inventory import (
CORRECT_JUDGMENT_ESSENCE,
CROSS_OVER_ESSENCE,
RITUAL_SUCCESS_ESSENCE,
)
from app.models.entity import Entity
from app.models.user import User
from app.rate_limit import RateLimiter
class FakeSpiritService:
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"):
return "listen"
async def wire_whisper(self, telemetry, language="en"):
return "the wire hums"
def chat_stream(self, entity, question, history, language="en"):
async def gen():
for token in ["I ", "am ", "here."]:
yield token
return gen()
def ambient_ready(self):
return False
async def _fake_synth(text, voice, profile, instability=0.0):
return b"RIFFfake wav bytes"
@pytest.fixture(autouse=True)
def _fake_spirits(monkeypatch):
monkeypatch.setattr(app.ws, "spirit_service", FakeSpiritService())
monkeypatch.setattr(app.ws, "synthesize_spirit_voice", _fake_synth)
# Generous, test-scoped limiters — the module-level ones are shared
# singletons that accumulate real hit counts across the whole test
# session (see backend/tests/test_ws_session.py's precedent), and this
# file summons repeatedly.
monkeypatch.setattr(app.ws, "summon_limiter", RateLimiter(max_requests=1000, window_seconds=60))
monkeypatch.setattr(app.ws, "summon_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60))
monkeypatch.setattr(app.ws, "question_limiter", RateLimiter(max_requests=1000, window_seconds=60))
monkeypatch.setattr(app.ws, "question_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60))
monkeypatch.setattr(app.ws, "fragment_limiter", RateLimiter(max_requests=1000, window_seconds=60))
monkeypatch.setattr(app.ws, "fragment_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60))
monkeypatch.setattr(app.ws, "ritual_limiter", RateLimiter(max_requests=1000, window_seconds=60))
monkeypatch.setattr(app.ws, "ritual_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60))
monkeypatch.setattr(app.ws, "judgment_limiter", RateLimiter(max_requests=1000, window_seconds=60))
monkeypatch.setattr(app.ws, "judgment_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60))
def _read_until(ws, msg_type, max_frames=60, **match):
for _ in range(max_frames):
frame = ws.receive_json()
if frame.get("type") != msg_type:
continue
if all(frame.get(key) == value for key, value in match.items()):
return frame
raise AssertionError(f"never saw frame of type {msg_type!r} matching {match!r}")
def _login(sync_client, username):
sync_client.post("/auth/register", json={"username": username, "password": "spookyspooky"})
sync_client.post("/auth/login", json={"username": username, "password": "spookyspooky"})
return sync_client.cookies.get("qm_session")
def _ws_connect(sync_client, token):
return sync_client.websocket_connect(
"/ws/session", headers={"cookie": f"qm_session={token}"}
)
def _summon(ws):
ws.send_json({"type": "summon"})
entity_frame = _read_until(ws, "entity")
_read_until(ws, "utterance", kind="greeting")
return entity_frame
def _run_ritual(ws, steps=4):
ws.send_json({"type": "ritual_start"})
for i in range(1, steps + 1):
ws.send_json({"type": "ritual_step", "step": i})
result = _read_until(ws, "ritual_complete")
# `_handle_ritual_step`'s reward call (essence credit / item roll) runs
# *after* the `ritual_complete` frame is queued for send, so receiving
# that frame doesn't by itself guarantee the reward has landed yet (the
# sender task drains the queue concurrently with the handler's own
# in-flight awaits). A trailing ping/pong forces a full round trip
# through the single connection's sequential message loop, which can't
# read the next message until the ritual_step handler (reward included)
# has fully returned — so seeing the pong is a hard guarantee, not a
# poll-and-hope.
ws.send_json({"type": "ping"})
_read_until(ws, "pong")
return result
# --- entity frame never leaks traits ---------------------------------------
@pytest.mark.asyncio
async def test_entity_frame_never_includes_traits(sync_client):
_login(sync_client, "no-leak")
with _ws_connect(sync_client, sync_client.cookies.get("qm_session")) as ws:
_read_until(ws, "session")
entity_frame = _summon(ws)
assert "traits" not in entity_frame["entity"]
# --- ritual ------------------------------------------------------------
@pytest.mark.asyncio
async def test_ritual_success_reveals_true_traits_and_credits_essence(
sync_client, db_session, monkeypatch
):
monkeypatch.setattr(app.ws.judgment, "roll_ritual_success", lambda traits, rng=None: True)
token = _login(sync_client, "ritual-winner")
user_id = uuid.UUID(
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
)
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
_summon(ws)
result = _run_ritual(ws)
assert result["success"] is True
assert result["revealed"] is not None
assert set(result["revealed"].keys()) == {
"alignment",
"power",
"volatility",
"deceptiveness",
}
for v in result["revealed"].values():
assert 0.0 <= v <= 1.0
# trickle (summon) + ritual success milestone. `_run_ritual`'s trailing
# ping/pong (see its docstring comment) guarantees the reward has fully
# landed before we get here.
from app.inventory import SUMMON_ESSENCE_TRICKLE
db_session.expire_all()
user = await db_session.get(User, user_id)
assert user.essence == SUMMON_ESSENCE_TRICKLE + RITUAL_SUCCESS_ESSENCE
@pytest.mark.asyncio
async def test_ritual_failure_reveals_nothing_and_grants_no_essence(
sync_client, db_session, monkeypatch
):
monkeypatch.setattr(app.ws.judgment, "roll_ritual_success", lambda traits, rng=None: False)
token = _login(sync_client, "ritual-loser")
user_id = uuid.UUID(
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
)
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
_summon(ws)
result = _run_ritual(ws)
assert result["success"] is False
assert result["revealed"] is None
db_session.expire_all()
user = await db_session.get(User, user_id)
from app.inventory import SUMMON_ESSENCE_TRICKLE
assert user.essence == SUMMON_ESSENCE_TRICKLE
@pytest.mark.asyncio
async def test_ritual_complete_only_fires_after_all_steps(sync_client, monkeypatch):
monkeypatch.setattr(app.ws.judgment, "roll_ritual_success", lambda traits, rng=None: True)
_login(sync_client, "ritual-partial")
with _ws_connect(sync_client, sync_client.cookies.get("qm_session")) as ws:
_read_until(ws, "session")
_summon(ws)
ws.send_json({"type": "ritual_start"})
ws.send_json({"type": "ritual_step", "step": 1})
ws.send_json({"type": "ritual_step", "step": 2})
ws.send_json({"type": "ping"})
pong = _read_until(ws, "pong")
assert pong == {"type": "pong"}
# No ritual_complete should have arrived yet (only 2/4 steps done) —
# if it had, it'd have been consumed as the "pong" read above
# skipped it via _read_until's scan, so assert explicitly by
# finishing the remaining steps and confirming exactly one
# ritual_complete follows.
ws.send_json({"type": "ritual_step", "step": 3})
ws.send_json({"type": "ritual_step", "step": 4})
result = _read_until(ws, "ritual_complete")
assert result["success"] is True
@pytest.mark.asyncio
async def test_fresh_summon_resets_ritual_progress(sync_client, monkeypatch):
monkeypatch.setattr(app.ws.judgment, "roll_ritual_success", lambda traits, rng=None: True)
_login(sync_client, "ritual-reset")
with _ws_connect(sync_client, sync_client.cookies.get("qm_session")) as ws:
_read_until(ws, "session")
_summon(ws)
ws.send_json({"type": "ritual_start"})
ws.send_json({"type": "ritual_step", "step": 1})
ws.send_json({"type": "ritual_step", "step": 2})
# A brand-new summon should discard the in-progress ritual — the
# next 4 steps on the new entity shouldn't complete after only 2
# more (i.e. carry over the old count).
_summon(ws)
ws.send_json({"type": "ritual_start"})
ws.send_json({"type": "ritual_step", "step": 1})
ws.send_json({"type": "ritual_step", "step": 2})
ws.send_json({"type": "ping"})
pong = _read_until(ws, "pong")
assert pong == {"type": "pong"}
# --- judgment: trust / banish -----------------------------------------------
@pytest.mark.asyncio
async def test_judgment_trust_correct_on_benevolent_entity(sync_client, db_session, monkeypatch):
monkeypatch.setattr(
app.ws.judgment,
"judge_verdict",
lambda verdict, traits, **kw: judgment_module.JudgmentOutcome(
True, 0.05, CORRECT_JUDGMENT_ESSENCE, False, "reward"
),
)
token = _login(sync_client, "truster")
user_id = uuid.UUID(
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
)
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
_summon(ws)
ws.send_json({"type": "judgment", "verdict": "trust"})
result = _read_until(ws, "judgment_result")
assert result == {
"type": "judgment_result",
"correct": True,
"favor_delta": 0.05,
"essence_delta": CORRECT_JUDGMENT_ESSENCE,
"at_peace": False,
"consequence": "reward",
}
db_session.expire_all()
user = await db_session.get(User, user_id)
assert user.favor == pytest.approx(0.05)
@pytest.mark.asyncio
async def test_judgment_wrong_trust_emits_escalation_consequence(sync_client, db_session, monkeypatch):
monkeypatch.setattr(
app.ws.judgment,
"judge_verdict",
lambda verdict, traits, **kw: judgment_module.JudgmentOutcome(
False, -0.10, 0, False, "escalation"
),
)
token = _login(sync_client, "wrong-truster")
user_id = uuid.UUID(
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
)
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
_summon(ws)
ws.send_json({"type": "judgment", "verdict": "trust"})
result = _read_until(ws, "judgment_result")
assert result["consequence"] == "escalation"
assert result["correct"] is False
assert result["favor_delta"] == -0.10
db_session.expire_all()
user = await db_session.get(User, user_id)
assert user.favor == pytest.approx(-0.10)
@pytest.mark.asyncio
async def test_judgment_favor_clamped_at_negative_one(sync_client, db_session, monkeypatch):
monkeypatch.setattr(
app.ws.judgment,
"judge_verdict",
lambda verdict, traits, **kw: judgment_module.JudgmentOutcome(
False, -0.10, 0, False, "escalation"
),
)
token = _login(sync_client, "favor-floor")
user_id = uuid.UUID(
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
)
async with app.ws.session_maker() as db:
user = await db.get(User, user_id)
user.favor = -0.95
await db.commit()
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
_summon(ws)
ws.send_json({"type": "judgment", "verdict": "trust"})
result = _read_until(ws, "judgment_result")
assert result["favor_delta"] == -0.10 # the raw per-event delta, unclamped
db_session.expire_all()
user = await db_session.get(User, user_id)
assert user.favor == pytest.approx(-1.0) # but the stored balance is clamped
@pytest.mark.asyncio
async def test_judgment_favor_clamped_at_positive_one(sync_client, db_session, monkeypatch):
monkeypatch.setattr(
app.ws.judgment,
"judge_verdict",
lambda verdict, traits, **kw: judgment_module.JudgmentOutcome(
True, 0.08, CROSS_OVER_ESSENCE, True, "crossed_over"
),
)
token = _login(sync_client, "favor-ceiling")
user_id = uuid.UUID(
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
)
async with app.ws.session_maker() as db:
user = await db.get(User, user_id)
user.favor = 0.97
await db.commit()
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
_summon(ws)
ws.send_json({"type": "judgment", "verdict": "cross_over"})
_read_until(ws, "judgment_result")
db_session.expire_all()
user = await db_session.get(User, user_id)
assert user.favor == pytest.approx(1.0)
# --- judgment: cross_over / at_peace ----------------------------------------
@pytest.mark.asyncio
async def test_judgment_cross_over_correct_sets_at_peace_and_pays_most_essence(
sync_client, db_session, monkeypatch
):
monkeypatch.setattr(
app.ws.judgment,
"judge_verdict",
lambda verdict, traits, **kw: judgment_module.JudgmentOutcome(
True, 0.08, CROSS_OVER_ESSENCE, True, "crossed_over"
),
)
token = _login(sync_client, "crosser")
user_id = uuid.UUID(
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
)
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
entity_frame = _summon(ws)
entity_id = uuid.UUID(entity_frame["entity"]["id"])
ws.send_json({"type": "judgment", "verdict": "cross_over"})
result = _read_until(ws, "judgment_result")
assert result["consequence"] == "crossed_over"
assert result["at_peace"] is True
assert result["essence_delta"] == CROSS_OVER_ESSENCE
db_session.expire_all()
entity = await db_session.get(Entity, entity_id)
assert entity.at_peace is True
from app.inventory import SUMMON_ESSENCE_TRICKLE
user = await db_session.get(User, user_id)
assert user.essence == SUMMON_ESSENCE_TRICKLE + CROSS_OVER_ESSENCE
@pytest.mark.asyncio
async def test_judgment_cross_over_resisted_on_demon_has_no_effect(sync_client, db_session, monkeypatch):
monkeypatch.setattr(
app.ws.judgment,
"judge_verdict",
lambda verdict, traits, **kw: judgment_module.JudgmentOutcome(
False, 0.0, 0, False, "resisted"
),
)
token = _login(sync_client, "resisted-demon")
user_id = uuid.UUID(
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
)
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
entity_frame = _summon(ws)
entity_id = uuid.UUID(entity_frame["entity"]["id"])
ws.send_json({"type": "judgment", "verdict": "cross_over"})
result = _read_until(ws, "judgment_result")
assert result == {
"type": "judgment_result",
"correct": False,
"favor_delta": 0.0,
"essence_delta": 0,
"at_peace": False,
"consequence": "resisted",
}
db_session.expire_all()
entity = await db_session.get(Entity, entity_id)
assert entity.at_peace is False
from app.inventory import SUMMON_ESSENCE_TRICKLE
user = await db_session.get(User, user_id)
assert user.essence == SUMMON_ESSENCE_TRICKLE
assert user.favor == 0.0
# --- judgment: test ----------------------------------------------------
@pytest.mark.asyncio
async def test_judgment_test_verdict_is_always_allowed_and_neutral(sync_client, db_session):
token = _login(sync_client, "tester")
user_id = uuid.UUID(
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
)
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
_summon(ws)
# No ritual attempted at all — should not crash, just no effect.
ws.send_json({"type": "judgment", "verdict": "test"})
result = _read_until(ws, "judgment_result")
assert result == {
"type": "judgment_result",
"correct": False,
"favor_delta": 0,
"essence_delta": 0,
"at_peace": False,
"consequence": "neutral",
}
db_session.expire_all()
user = await db_session.get(User, user_id)
assert user.favor == 0.0
@pytest.mark.asyncio
async def test_judgment_test_verdict_correct_after_successful_ritual(sync_client, monkeypatch):
monkeypatch.setattr(app.ws.judgment, "roll_ritual_success", lambda traits, rng=None: True)
_login(sync_client, "tester-after-ritual")
with _ws_connect(sync_client, sync_client.cookies.get("qm_session")) as ws:
_read_until(ws, "session")
_summon(ws)
_run_ritual(ws)
ws.send_json({"type": "judgment", "verdict": "test"})
result = _read_until(ws, "judgment_result")
assert result["correct"] is True
assert result["consequence"] == "neutral"
assert result["favor_delta"] == 0
assert result["essence_delta"] == 0
# --- at_peace + re-contact ----------------------------------------------
@pytest.mark.asyncio
async def test_at_peace_entity_is_not_recontacted_a_fresh_one_mints_instead(
sync_client, db_session
):
anomalies = [
{"type": "anomaly", "source": "radio", "frequency": 101.0 + i, "magnitude": 5.0 + i}
for i in range(4)
]
token = _login(sync_client, "peace-seeker")
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
for anomaly in anomalies:
ws.send_json(anomaly)
first_frame = _read_until(ws, "entity")
first_id = uuid.UUID(first_frame["entity"]["id"])
first_signature = None # not exposed to the client; fetched below
# Manually mark the entity at_peace, as a completed cross_over would.
async with app.ws.session_maker() as db:
entity = await db.get(Entity, first_id)
entity.at_peace = True
first_signature = entity.signature
await db.commit()
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
for anomaly in anomalies:
ws.send_json(anomaly)
second_frame = _read_until(ws, "entity")
assert second_frame["is_new"] is True
second_id = uuid.UUID(second_frame["entity"]["id"])
assert second_id != first_id
db_session.expire_all()
second_entity = await db_session.get(Entity, second_id)
assert second_entity.at_peace is False
# Salted variant of the same base signature, not a raw collision.
assert second_entity.signature != first_signature
assert second_entity.signature.startswith(first_signature + ":")
first_entity = await db_session.get(Entity, first_id)
assert first_entity is not None # memorialized, never deleted
assert first_entity.at_peace is True
# --- favor read-back bias on trait rolls at mint time -----------------------
@pytest.mark.asyncio
async def test_high_favor_user_mints_less_volatile_deceptive_entities(sync_client, db_session):
token = _login(sync_client, "high-favor-summoner")
user_id = uuid.UUID(
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
)
async with app.ws.session_maker() as db:
user = await db.get(User, user_id)
user.favor = 1.0
await db.commit()
anomalies = [
{"type": "anomaly", "source": "radio", "frequency": 201.0 + i, "magnitude": 5.0 + i}
for i in range(4)
]
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
for anomaly in anomalies:
ws.send_json(anomaly)
entity_frame = _read_until(ws, "entity")
entity_id = uuid.UUID(entity_frame["entity"]["id"])
db_session.expire_all()
entity = await db_session.get(Entity, entity_id)
# Recompute what the unbiased roll would have been for this signature
# and confirm the stored traits were nudged toward more legible
# (lower volatility/deceptiveness), never the other direction.
from app.entities import roll_traits
unbiased = roll_traits(entity.signature)
assert entity.traits["volatility"] <= unbiased["volatility"]
assert entity.traits["deceptiveness"] <= unbiased["deceptiveness"]
# alignment/power are untouched by the bias.
assert entity.traits["alignment"] == pytest.approx(unbiased["alignment"])
assert entity.traits["power"] == pytest.approx(unbiased["power"])
@pytest.mark.asyncio
async def test_low_favor_user_mints_more_volatile_deceptive_entities(sync_client, db_session):
token = _login(sync_client, "low-favor-summoner")
user_id = uuid.UUID(
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
)
async with app.ws.session_maker() as db:
user = await db.get(User, user_id)
user.favor = -1.0
await db.commit()
anomalies = [
{"type": "anomaly", "source": "radio", "frequency": 301.0 + i, "magnitude": 5.0 + i}
for i in range(4)
]
with _ws_connect(sync_client, token) as ws:
_read_until(ws, "session")
for anomaly in anomalies:
ws.send_json(anomaly)
entity_frame = _read_until(ws, "entity")
entity_id = uuid.UUID(entity_frame["entity"]["id"])
db_session.expire_all()
entity = await db_session.get(Entity, entity_id)
from app.entities import roll_traits
unbiased = roll_traits(entity.signature)
assert entity.traits["volatility"] >= unbiased["volatility"]
assert entity.traits["deceptiveness"] >= unbiased["deceptiveness"]
Reference in New Issue View Git Blame Copy Permalink