From 36c4a9e6e4a014258b25075509139dc8488e5dc8 Mon Sep 17 00:00:00 2001 From: Indiana Date: Fri, 24 Jul 2026 21:27:30 +0000 Subject: [PATCH] feat: ritual + judgment + favor + cross-over (Workstream B) Implements Workstream B of the character-depth-ghost-log spec: ritual_start/ritual_step/judgment WS handlers, the pure judgment.py logic module, and the User.favor / Entity.at_peace columns + migration. - app/judgment.py: pure ritual success roll (base 65%, floored at 30%, driven by an entity's power+deceptiveness difficulty), the "stuck spirit" cross_over rule (alignment >= 0.5 and volatility > 0.6, ~20% of entities), judgment correctness/favor-delta/essence-delta/ consequence resolution for all four verdicts, favor clamping, the favor-to-trait-roll bias applied at mint time, and tell-line generation (opaque behavioral flavor text, never a raw stat). - app/ws.py: wires ritual_start/ritual_step/judgment frames, emits ritual_complete/tell/judgment_result/item_drop per the spec's Contract; traits are added to serialize_entity for internal server-side use but stripped from the outbound `entity` frame via a new _public_entity helper so hidden ground truth never reaches the client outside ritual_complete; _summon excludes at-peace entities from signature re-contact and mints a fresh (salted-signature) entity instead; new entities' traits are nudged by the discovering user's favor before being persisted. - models/user.py, models/entity.py, main.py: User.favor and Entity.at_peace columns plus their idempotent ADD COLUMN IF NOT EXISTS migration lines in lifespan, alongside the existing ones. - tests/test_judgment.py, tests/test_ws_ritual_judgment.py: 56 new tests covering the ritual/judgment correctness matrix, favor clamping/bias, essence crediting, at_peace persistence + re-contact, and the entity-frame trait leak guard. Co-Authored-By: Claude Sonnet 5 --- backend/app/judgment.py | 315 ++++++++++++ backend/app/main.py | 7 + backend/app/models/entity.py | 8 +- backend/app/models/user.py | 13 +- backend/app/ws.py | 240 ++++++++- backend/tests/test_judgment.py | 363 ++++++++++++++ backend/tests/test_ws_ritual_judgment.py | 614 +++++++++++++++++++++++ 7 files changed, 1541 insertions(+), 19 deletions(-) create mode 100644 backend/app/judgment.py create mode 100644 backend/tests/test_judgment.py create mode 100644 backend/tests/test_ws_ritual_judgment.py diff --git a/backend/app/judgment.py b/backend/app/judgment.py new file mode 100644 index 0000000..a91a942 --- /dev/null +++ b/backend/app/judgment.py @@ -0,0 +1,315 @@ +"""Ritual + judgment: pure logic for Workstream B of the Character Depth / +Ghost Log spec +(docs/superpowers/specs/2026-07-23-character-depth-ghost-log-design.md). + +No I/O, no DB, no network — every function here takes plain values (and an +optional injectable `random.Random`) and returns plain values, so the whole +module is trivially unit-testable in isolation. `backend/app/ws.py`'s +`ritual_start`/`ritual_step`/`judgment` WS handlers are the only place these +outputs get persisted or sent over the wire. +""" + +from __future__ import annotations + +import random +from dataclasses import dataclass + +from app.inventory import CORRECT_JUDGMENT_ESSENCE, CROSS_OVER_ESSENCE + +# --- favor ------------------------------------------------------------- + +FAVOR_MIN = -1.0 +FAVOR_MAX = 1.0 + + +def clamp_favor(value: float) -> float: + """Clamps a `User.favor` value to [-1.0, 1.0] — call this at every write + site, per the spec's Contract section.""" + return max(FAVOR_MIN, min(FAVOR_MAX, value)) + + +# --- ritual -------------------------------------------------------------- + +# Rituals should feel doable most of the time — this is a short rite in +# front of judgment, not a grindy gate. A highly resistant entity (high +# `power` *and* high `deceptiveness` — strong and cagey) drags the odds +# down, floored so even the worst-case entity stays beatable on a retry +# rather than a hard wall. +RITUAL_BASE_SUCCESS_CHANCE = 0.65 +RITUAL_MIN_SUCCESS_CHANCE = 0.30 +RITUAL_DIFFICULTY_SWING = RITUAL_BASE_SUCCESS_CHANCE - RITUAL_MIN_SUCCESS_CHANCE + + +def roll_ritual_success(traits: dict, rng: random.Random | None = None) -> bool: + """One ritual attempt's pass/fail roll. + + `traits` is the entity's hidden truth dict (`alignment`/`power`/ + `volatility`/`deceptiveness`). Only `power` and `deceptiveness` affect + the odds — a strong, cagey spirit is the hardest to get an accurate + read on. `volatility` is deliberately left out: it's about how *noisy* + the entity's ambient tells are, a separate axis from whether a focused + ritual can pin it down. `alignment` is also left out — letting it + influence pass/fail would telegraph ground truth (easier ritual = + probably benevolent) before `ritual_complete` is supposed to reveal + anything, undermining the "judge blind or judge informed" choice the + contract is built around. + + `rng` defaults to a fresh, unseeded `random.Random()` — deliberately + *not* signature-seeded like `entities.roll_traits`: a seeker can retry + the same entity's ritual repeatedly (`RitualPanel`'s "attempt again"), + and each attempt needs genuine variance, not a fixed pass/fail baked + into the entity's identity. + """ + rng = rng if rng is not None else random.Random() + power = float(traits.get("power", 0.5)) + deceptiveness = float(traits.get("deceptiveness", 0.5)) + difficulty = (power + deceptiveness) / 2.0 # 0..1 + chance = RITUAL_BASE_SUCCESS_CHANCE - RITUAL_DIFFICULTY_SWING * difficulty + chance = max(RITUAL_MIN_SUCCESS_CHANCE, min(RITUAL_BASE_SUCCESS_CHANCE, chance)) + return rng.random() < chance + + +# --- "stuck spirit" rule for cross_over ----------------------------------- + +# "Genuinely benevolent" mirrors the correct-trust threshold used below. +# "Reads as stuck" is modeled as above-average volatility: the existing +# fallback personas already lean on "died with something unfinished" themes +# (see entities.py's _PERSONA_TEMPLATES), and an unresolved spirit is +# exactly the kind whose behavioral tells would plausibly be noisy/ +# inconsistent rather than settled, rather than tying "stuck" to a whole +# separate hidden dimension. 0.6 (vs. the 0.5 midpoint) keeps "stuck" a +# meaningfully above-average band rather than a coinflip: with both +# `alignment` and `volatility` rolled uniform(0, 1) independently, +# P(alignment >= 0.5) * P(volatility > 0.6) = 0.5 * 0.4 = 20% of all +# entities qualify — special enough to justify the single largest reward +# in the game (see CROSS_OVER_ESSENCE / FAVOR_CORRECT_CROSS_OVER) without +# being so rare it never comes up. +BENEVOLENT_ALIGNMENT_THRESHOLD = 0.5 +STUCK_VOLATILITY_THRESHOLD = 0.6 + + +def is_stuck_spirit(traits: dict) -> bool: + alignment = float(traits.get("alignment", 0.5)) + volatility = float(traits.get("volatility", 0.5)) + return ( + alignment >= BENEVOLENT_ALIGNMENT_THRESHOLD + and volatility > STUCK_VOLATILITY_THRESHOLD + ) + + +# --- judgment -------------------------------------------------------------- + +# Favor deltas: small, single-digit-percent nudges of the [-1, 1] range (a +# 2.0-wide band) — the caller always re-clamps via `clamp_favor`. +# +# Wrong-trust is punished harder than wrong-banish: trusting something that +# turns out malevolent is the reckless failure mode with real in-fiction +# consequences (the haunting escalates), while a wrong banish is merely +# over-cautious — you turned away something harmless, nothing lashes back. +# Correct cross_over pays the best favor *and* essence of any outcome, +# matching the contract's "largest essence reward of any outcome" language +# with an equivalently generous favor nudge: it's the hardest-to-spot, +# most compassionate correct call a seeker can make. +FAVOR_CORRECT_TRUST = 0.05 +FAVOR_CORRECT_BANISH = 0.05 +FAVOR_WRONG_TRUST = -0.10 +FAVOR_WRONG_BANISH = -0.05 +FAVOR_CORRECT_CROSS_OVER = 0.08 +FAVOR_CROSS_OVER_FAIL = 0.0 # a naive read, not a reckless one — no penalty +FAVOR_TEST = 0.0 # a diagnostic pulse only — nothing risked, nothing gained + +VERDICTS = ("trust", "banish", "test", "cross_over") + + +@dataclass(frozen=True) +class JudgmentOutcome: + correct: bool + favor_delta: float + essence_delta: int + at_peace: bool + consequence: str # "reward" | "escalation" | "withdrawal" | "crossed_over" | "resisted" | "neutral" + + +def judge_verdict( + verdict: str, + traits: dict, + *, + ritual_completed: bool = False, + ritual_success: bool = False, +) -> JudgmentOutcome: + """Resolves one `judgment` frame against the entity's hidden truth. + + `correct` per the contract: trust called on real alignment >= 0.5, or + banish called on alignment < 0.5, or cross_over called on a genuinely + "stuck" spirit (see `is_stuck_spirit`). `cross_over` on anything else + (a demon, or a benevolent-but-not-stuck spirit that simply isn't ready + to move on) resists — "resisted" covers both: neither is a reckless + misread the way a wrong trust/banish is, so neither costs favor. + `test` never touches favor/essence; its `correct` reflects whether a + completed-this-session ritual actually surfaced true information (no + completed ritual, or a failed one, means the diagnostic has nothing + real to go on). + """ + if verdict not in VERDICTS: + raise ValueError(f"unknown verdict: {verdict!r}") + + alignment = float(traits.get("alignment", 0.5)) + benevolent = alignment >= BENEVOLENT_ALIGNMENT_THRESHOLD + + if verdict == "trust": + if benevolent: + return JudgmentOutcome( + True, FAVOR_CORRECT_TRUST, CORRECT_JUDGMENT_ESSENCE, False, "reward" + ) + return JudgmentOutcome(False, FAVOR_WRONG_TRUST, 0, False, "escalation") + + if verdict == "banish": + if not benevolent: + return JudgmentOutcome( + True, FAVOR_CORRECT_BANISH, CORRECT_JUDGMENT_ESSENCE, False, "reward" + ) + return JudgmentOutcome(False, FAVOR_WRONG_BANISH, 0, False, "withdrawal") + + if verdict == "cross_over": + if is_stuck_spirit(traits): + return JudgmentOutcome( + True, FAVOR_CORRECT_CROSS_OVER, CROSS_OVER_ESSENCE, True, "crossed_over" + ) + return JudgmentOutcome(False, FAVOR_CROSS_OVER_FAIL, 0, False, "resisted") + + # verdict == "test" + correct = bool(ritual_completed and ritual_success) + return JudgmentOutcome(correct, FAVOR_TEST, 0, False, "neutral") + + +# --- favor read-back bias on trait rolls ------------------------------------ + +# `entities.roll_traits` stays a pure, signature-seeded function with no +# session/user awareness (Workstream A's design — see the "roll_traits +# doesn't take a bias/rng parameter" check in ws.py's minting path). This is +# applied instead as a post-processing nudge, called from `app.ws._summon` +# right after a *new* entity's profile is minted, using the discovering +# user's current favor. +# +# It's applied once, at mint time, not per-viewing-session: the nudged +# traits become that entity's permanent, shared ground truth in the Codex +# (the same spirit reads the same way to every future seeker who contacts +# it). That matches the contract's "read back to mildly bias future summon +# trait rolls" framing — favor shapes what a seeker tends to *conjure*, +# not a private lens they view existing spirits through. +# +# Effect size is deliberately small and verifiable: at most a 0.12 shift at +# |favor| == 1.0, linear in between, applied only to volatility/ +# deceptiveness (the two "how legible is this entity" axes) — alignment and +# power are left untouched so favor nudges readability, not who a spirit +# fundamentally is. +FAVOR_TRAIT_BIAS_MAX = 0.12 +_BIASED_TRAIT_KEYS = ("volatility", "deceptiveness") + + +def apply_favor_bias(traits: dict, favor: float) -> dict: + """Higher favor nudges volatility/deceptiveness down (more legible + entities); lower favor nudges them up. Returns a new dict; clamps each + nudged value back into [0.0, 1.0].""" + favor = clamp_favor(favor) + shift = -FAVOR_TRAIT_BIAS_MAX * favor + biased = dict(traits) + for key in _BIASED_TRAIT_KEYS: + if key in biased: + biased[key] = max(0.0, min(1.0, float(biased[key]) + shift)) + return biased + + +# --- tells ----------------------------------------------------------------- + +# Flavor text describing *behavior*, never a stat number (per the contract +# and frontend/src/lib/evilMeter.ts's comments on how tells are consumed) — +# each line hints at one trait dimension without naming it. Picking one +# trait per call (rather than always the most extreme) keeps tells varied +# session to session even for the same entity; the high/low/neutral banding +# means a middling trait still produces a plausible, non-committal line +# instead of always screaming its most extreme dimension. +_TELL_LINES: dict[str, dict[str, tuple[str, ...]]] = { + "alignment": { + "high": ( + "a warmth threads through the static, unmistakably kind.", + "it seems to want nothing more than to be heard.", + "the presence feels gentle, almost grateful for the company.", + ), + "low": ( + "something cold coils under the words.", + "you feel watched, not accompanied.", + "the air around the signal turns unfriendly.", + ), + "neutral": ( + "hard to say if it means well.", + "the intent behind it stays unreadable.", + ), + }, + "power": { + "high": ( + "the signal surges, straining the line.", + "it pushes back against the questions, strong-willed.", + ), + "low": ( + "the presence feels thin, easily startled.", + "it flickers at the edge of hearing.", + ), + "neutral": ( + "steady, unremarkable strength.", + "neither weak nor overwhelming.", + ), + }, + "volatility": { + "high": ( + "the tone lurches without warning.", + "it contradicts itself within the same breath.", + "the signal keeps slipping out from under itself.", + ), + "low": ( + "calm, consistent, almost rehearsed.", + "every answer lands the same measured way.", + ), + "neutral": ( + "mostly steady, with the odd hitch.", + "a little uneven, nothing alarming.", + ), + }, + "deceptiveness": { + "high": ( + "the entity avoided a direct question.", + "an answer arrives that doesn't quite fit what was asked.", + "something about the reply feels rehearsed, not remembered.", + ), + "low": ( + "it answers plainly, almost bluntly.", + "nothing about the reply feels rehearsed.", + ), + "neutral": ( + "the reply seems straightforward enough.", + "no obvious dodge, no obvious tell.", + ), + }, +} + +_TELL_TRAIT_KEYS = tuple(_TELL_LINES.keys()) +TELL_HIGH_THRESHOLD = 0.6 +TELL_LOW_THRESHOLD = 0.4 + + +def generate_tell(traits: dict, rng: random.Random) -> str: + """Picks one trait dimension at random and returns a short, opaque + flavor line hinting at it (never the raw number). `rng` should be a + per-session `random.Random` so a given session's tell sequence is + varied but the choice of *which* call sites emit a tell stays the + caller's (ws.py's) responsibility.""" + trait_key = rng.choice(_TELL_TRAIT_KEYS) + value = float(traits.get(trait_key, 0.5)) + bank = _TELL_LINES[trait_key] + if value >= TELL_HIGH_THRESHOLD: + pool = bank["high"] + elif value <= TELL_LOW_THRESHOLD: + pool = bank["low"] + else: + pool = bank["neutral"] + return rng.choice(pool) diff --git a/backend/app/main.py b/backend/app/main.py index 4cc8640..1ccbcf1 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -54,6 +54,13 @@ async def lifespan(app: FastAPI): await conn.execute(text( "ALTER TABLE entities ADD COLUMN IF NOT EXISTS traits JSONB NOT NULL DEFAULT '{}'::jsonb" )) + # Workstream B (character-depth-ghost-log spec). + await conn.execute(text( + "ALTER TABLE users ADD COLUMN IF NOT EXISTS favor DOUBLE PRECISION NOT NULL DEFAULT 0.0" + )) + await conn.execute(text( + "ALTER TABLE entities ADD COLUMN IF NOT EXISTS at_peace BOOLEAN NOT NULL DEFAULT false" + )) cleanup_task = asyncio.create_task(_session_cleanup_loop()) try: yield diff --git a/backend/app/models/entity.py b/backend/app/models/entity.py index 5a18f61..6a9c6ca 100644 --- a/backend/app/models/entity.py +++ b/backend/app/models/entity.py @@ -1,7 +1,7 @@ import uuid from datetime import datetime, timezone -from sqlalchemy import DateTime, ForeignKey, Integer, String, Text +from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column @@ -27,6 +27,12 @@ class Entity(Base): traits: Mapped[dict] = mapped_column(JSONB, default=dict) sample_quotes: Mapped[list] = mapped_column(JSONB, default=list) contact_count: Mapped[int] = mapped_column(Integer, default=0) + # Workstream B (character-depth-ghost-log spec): set true when a seeker + # correctly helps a genuinely benevolent, "stuck" spirit cross over. The + # row is never deleted (memorialized in the Codex permanently) but + # `signature_from_anomalies` re-contact in app.ws._summon skips it and + # mints a fresh entity instead. + at_peace: Mapped[bool] = mapped_column(Boolean, default=False) discovered_by: Mapped[uuid.UUID | None] = mapped_column( ForeignKey("users.id"), nullable=True ) diff --git a/backend/app/models/user.py b/backend/app/models/user.py index 33e0a92..4faae46 100644 --- a/backend/app/models/user.py +++ b/backend/app/models/user.py @@ -1,7 +1,7 @@ import uuid from datetime import datetime, timezone -from sqlalchemy import DateTime, Integer, String +from sqlalchemy import DateTime, Float, Integer, String from sqlalchemy.orm import Mapped, mapped_column from app.db import Base @@ -14,13 +14,12 @@ class User(Base): username: Mapped[str] = mapped_column(String(32), unique=True, index=True) password_hash: Mapped[str] = mapped_column(String(255)) email: Mapped[str | None] = mapped_column(String(255), nullable=True) - # NOTE: owned by Workstream B in the character-depth-ghost-log spec - # (docs/superpowers/specs/2026-07-23-character-depth-ghost-log-design.md). - # Added here so Workstream C (unlocks/items/sigils/drops) can build and - # test against it in isolation; the merge controller reconciles this - # against Workstream B's own edit to this file (which will also add - # `favor: float` per the spec's Contract section). essence: Mapped[int] = mapped_column(Integer, default=0) + # Workstream B (character-depth-ghost-log spec): hidden per-user score, + # nudged by judgment correctness, clamped to [-1.0, 1.0] everywhere it's + # written (see app.judgment.clamp_favor). Read back as a small bias on + # new entities' trait rolls at mint time (app.judgment.apply_favor_bias). + favor: Mapped[float] = mapped_column(Float, default=0.0) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=lambda: datetime.now(timezone.utc) ) diff --git a/backend/app/ws.py b/backend/app/ws.py index bc28634..01cc365 100644 --- a/backend/app/ws.py +++ b/backend/app/ws.py @@ -8,6 +8,10 @@ Protocol (client → server): {"type": "anomaly", "source": SRC, ...} → {"type": "utterance", ...} {"type": "question", "text": "..."} → reply_start / reply_token* / reply_end {"type": "passive", "enabled": bool} → ambient wire loop on/off + {"type": "ritual_start"} → (begins a ritual attempt) + {"type": "ritual_step", "step": } → (on the final step) ritual_complete + {"type": "judgment", "verdict": "trust" | "banish" | "test" | "cross_over"} + → judgment_result All server → client frames flow through a single sender task so concurrent producers (ambient loop, reply streaming, TTS callbacks) never interleave on @@ -26,11 +30,18 @@ from pathlib import Path from fastapi import APIRouter, WebSocket, WebSocketDisconnect from sqlalchemy import select +from app import judgment from app.config import settings from app.db import async_session_maker as _default_session_maker from app.deps import SESSION_COOKIE_NAME from app.entities import fallback_signature, signature_from_anomalies -from app.inventory import SUMMON_ESSENCE_TRICKLE, credit_essence, roll_item_drop, summon_drop_trigger +from app.inventory import ( + RITUAL_SUCCESS_ESSENCE, + SUMMON_ESSENCE_TRICKLE, + credit_essence, + roll_item_drop, + summon_drop_trigger, +) from app.llm.service import SpiritBusyError, spirit_service from app.models.auth_session import AuthSession, hash_token from app.models.contact_session import ContactSession @@ -89,6 +100,16 @@ class SeanceState: ambient_task: asyncio.Task | None = None wire_jitter_history: list[float] = field(default_factory=list) last_wire_anomaly_at: float = 0.0 + # Workstream B (character-depth-ghost-log spec): ritual progress for the + # *current* entity — reset whenever a fresh presence is summoned (see + # _handle_summon) or a new ritual_start arrives. + ritual_steps: int = 0 + ritual_completed: bool = False + ritual_success: bool = False + # Per-session RNG for `tell` frames — unseeded (a session's tells should + # vary run to run), but persistent across calls so the draw sequence + # isn't restarted on every single message. + tell_rng: random.Random = field(default_factory=random.Random) # Active-session registry (spec: ESP32 sensor node, Workstream K): maps a @@ -133,9 +154,24 @@ def serialize_entity(entity: Entity) -> dict: "quotes": entity.sample_quotes, "contact_count": entity.contact_count, "discovered_at": entity.discovered_at.isoformat(), + # Workstream B: hidden ground truth, kept on the server-side + # SeanceState.entity dict for the ritual/judgment/tell handlers to + # read (state.entity["traits"]) — see `_public_entity` below for why + # this never reaches the wire directly. + "traits": entity.traits, } +def _public_entity(entity: dict) -> dict: + """The entity payload actually sent to the client in the `entity` + frame — everything `serialize_entity` produces *except* `traits`. + Hidden traits must never leak outside `ritual_complete` on success (the + contract's decoupling requirement, echoed in + frontend/src/lib/evilMeter.ts's comments); `frontend/src/lib/types.ts`'s + `SpiritEntity` type correspondingly has no `traits` field.""" + return {key: value for key, value in entity.items() if key != "traits"} + + def _client_ip(websocket: WebSocket) -> str: host = websocket.client.host if websocket.client else None return resolve_client_ip(websocket.headers, host) @@ -237,28 +273,50 @@ async def _unique_entity_name(db, base_name: str) -> str: async def _summon(state: SeanceState, channel: str) -> tuple[Entity, bool]: - """Match this session's signature against the Codex, or mint a new entity.""" + """Match this session's signature against the Codex, or mint a new entity. + + An at-peace entity (Workstream B: a spirit correctly helped to cross + over) is excluded from the match — it stays in the Codex forever but + can't be re-contacted. If its signature is what this session's anomaly + pattern hashes to, a *new* entity is minted instead. `Entity.signature` + is unique, so the new entity can't reuse the exact same string while the + retired row still holds it — it gets a salted variant of the same base + signature instead. + """ signature = signature_from_anomalies(state.anomalies) or fallback_signature( str(state.session_id) ) async with session_maker() as db: - entity = await db.scalar(select(Entity).where(Entity.signature == signature)) + entity = await db.scalar( + select(Entity).where(Entity.signature == signature, Entity.at_peace.is_(False)) + ) is_new = entity is None if is_new: + mint_signature = signature + retired = await db.scalar(select(Entity).where(Entity.signature == signature)) + if retired is not None: + mint_signature = f"{signature}:{uuid.uuid4().hex[:8]}" + profile = await spirit_service.mint_profile( - signature, channel, state.anomalies, state.language + mint_signature, channel, state.anomalies, state.language ) + discoverer = await db.get(User, state.user_id) + favor = discoverer.favor if discoverer is not None else 0.0 entity = Entity( name=await _unique_entity_name(db, profile["name"]), epithet=profile["epithet"], persona=profile["persona"], rarity_tier=profile["rarity"], - signature=signature, + signature=mint_signature, voice_profile=profile["voice"], visual_profile=profile["visual"], sample_quotes=profile["quotes"], + # Workstream B: signature-seeded traits, nudged by the + # discovering user's favor (app.judgment.apply_favor_bias) — + # never derived from/fed into the persona above. + traits=judgment.apply_favor_bias(profile["traits"], favor), discovered_by=state.user_id, contact_count=1, ) @@ -286,11 +344,11 @@ async def _reward_summon(state: SeanceState) -> None: summon (any mode), and — only when the summoned entity is high-rarity — a roll for an item drop. The other two contract trigger points ("after a correct judgment, a successful ritual") belong to Workstream B's - ritual/judgment WS handlers, which don't exist in this codebase yet; - `app.inventory` exposes the same `roll_item_drop`/`credit_essence` - helpers (plus the milestone essence constants) for those handlers to - call once they land, so the drop table and essence economy stay in one - place instead of being duplicated.""" + ritual/judgment WS handlers (`_reward_ritual_success` / `_handle_judgment` + below), which call the same `app.inventory` `roll_item_drop`/ + `credit_essence` helpers and milestone essence constants so the drop + table and essence economy stay in one place instead of being + duplicated.""" assert state.entity is not None rarity = state.entity.get("rarity", "common") @@ -338,14 +396,39 @@ async def _handle_summon(state: SeanceState) -> None: await state.send_queue.put({"type": "status", "state": "summoning"}) entity, is_new = await _summon(state, state.mode if state.mode != "unknown" else "ouija") state.entity = serialize_entity(entity) + # A fresh presence invalidates any in-progress/completed ritual from + # whatever was previously in this slot (mirrors the frontend reducer's + # 'entity' case in state/seance.tsx, which resets its own ritual/ + # judgment UI state the same way). + state.ritual_steps = 0 + state.ritual_completed = False + state.ritual_success = False await state.send_queue.put( - {"type": "entity", "entity": state.entity, "is_new": is_new} + {"type": "entity", "entity": _public_entity(state.entity), "is_new": is_new} ) await _reward_summon(state) greeting = random.choice(state.entity["quotes"]) if state.entity["quotes"] else "I am here." await _speak(state, "greeting", greeting) +# Workstream B: `tell` frames piggyback on the existing anomaly/reply +# handling rather than running their own timer — a fragment (ambient, +# frequent) rolls a lower chance than a direct reply (deliberate, a seeker +# just asked something), so tells feel like they're punctuating engagement +# rather than firing on a fixed clock. +TELL_CHANCE_ON_FRAGMENT = 0.2 +TELL_CHANCE_ON_REPLY = 0.35 + + +async def _maybe_tell(state: SeanceState, chance: float) -> None: + if state.entity is None: + return + if state.tell_rng.random() >= chance: + return + text = judgment.generate_tell(state.entity.get("traits", {}), state.tell_rng) + await state.send_queue.put({"type": "tell", "text": text}) + + async def _handle_anomaly(state: SeanceState, message: dict) -> None: anomaly = { "source": str(message.get("source", "unknown"))[:16], @@ -378,6 +461,7 @@ async def _handle_anomaly(state: SeanceState, message: dict) -> None: except SpiritBusyError: return await _speak(state, "fragment", fragment) + await _maybe_tell(state, TELL_CHANCE_ON_FRAGMENT) async def _handle_question(state: SeanceState, text: str) -> None: @@ -434,6 +518,7 @@ async def _handle_question(state: SeanceState, text: str) -> None: await state.send_queue.put({"type": "reply_end", "id": str(reply_id), "text": reply}) if reply: await _speak(state, "reply", reply, instability=1 - stability) + await _maybe_tell(state, TELL_CHANCE_ON_REPLY) async def _ambient_loop(state: SeanceState) -> None: @@ -485,6 +570,133 @@ async def _handle_passive(state: SeanceState, enabled: bool) -> None: await state.send_queue.put({"type": "passive", "enabled": False}) +# --- Workstream B: ritual + judgment (character-depth-ghost-log spec) ------ + +# How many `ritual_step` frames complete one attempt — matches +# frontend/src/lib/ritual.ts's RITUAL_TOTAL_STEPS (the 4-rune "align / +# breathe / trace / lock" sequence). The frontend owns the exact step count +# per the spec ("implementer's call"); this just has to agree with it. +RITUAL_STEPS_REQUIRED = 4 + + +async def _handle_ritual_start(state: SeanceState) -> None: + if state.entity is None: + return # no presence to focus on — frontend already gates the button + state.ritual_steps = 0 + state.ritual_completed = False + state.ritual_success = False + + +async def _reward_ritual_success(state: SeanceState) -> None: + """Mirrors `_reward_summon`'s essence-credit + item-drop pattern for the + ritual milestone trigger point.""" + item = None + async with session_maker() as db: + user = await db.get(User, state.user_id) + if user is None: + return + credit_essence(user, RITUAL_SUCCESS_ESSENCE) + item = roll_item_drop("ritual") + if item is not None: + db.add( + InventoryItem( + user_id=state.user_id, + item_type=item["item_type"], + item_key=item["item_key"], + payload=item["payload"], + ) + ) + await db.commit() + + if item is not None: + await state.send_queue.put({"type": "item_drop", "item": item}) + + +async def _handle_ritual_step(state: SeanceState, message: dict) -> None: + if state.entity is None or state.ritual_completed: + return + if not isinstance(message.get("step"), int): + return + + state.ritual_steps += 1 + if state.ritual_steps < RITUAL_STEPS_REQUIRED: + return + + traits = state.entity.get("traits", {}) + success = judgment.roll_ritual_success(traits) + state.ritual_completed = True + state.ritual_success = success + revealed = dict(traits) if success else None + await state.send_queue.put( + {"type": "ritual_complete", "success": success, "revealed": revealed} + ) + if success: + await _reward_ritual_success(state) + + +async def _handle_judgment(state: SeanceState, message: dict) -> None: + if state.entity is None: + return + verdict = message.get("verdict") + if verdict not in judgment.VERDICTS: + return + + traits = state.entity.get("traits", {}) + outcome = judgment.judge_verdict( + verdict, + traits, + ritual_completed=state.ritual_completed, + ritual_success=state.ritual_success, + ) + + item = None + # Skip the DB round-trip entirely when there's nothing to persist (e.g. + # `test` without a completed ritual, or a resisted cross_over) — the + # contract's "no crash, just no effect" for those cases. + if outcome.favor_delta or outcome.essence_delta or outcome.consequence in ( + "reward", + "crossed_over", + ): + async with session_maker() as db: + user = await db.get(User, state.user_id) + if user is not None: + if outcome.favor_delta: + user.favor = judgment.clamp_favor(user.favor + outcome.favor_delta) + if outcome.essence_delta: + credit_essence(user, outcome.essence_delta) + + if outcome.consequence == "crossed_over": + entity_row = await db.get(Entity, uuid.UUID(state.entity["id"])) + if entity_row is not None: + entity_row.at_peace = True + + if outcome.consequence in ("reward", "crossed_over"): + item = roll_item_drop("judgment") + if item is not None: + db.add( + InventoryItem( + user_id=state.user_id, + item_type=item["item_type"], + item_key=item["item_key"], + payload=item["payload"], + ) + ) + await db.commit() + + await state.send_queue.put( + { + "type": "judgment_result", + "correct": outcome.correct, + "favor_delta": outcome.favor_delta, + "essence_delta": outcome.essence_delta, + "at_peace": outcome.at_peace, + "consequence": outcome.consequence, + } + ) + if item is not None: + await state.send_queue.put({"type": "item_drop", "item": item}) + + @router.websocket("/ws/session") async def session_socket(websocket: WebSocket) -> None: user_id = await _authenticate(websocket) @@ -539,6 +751,12 @@ async def session_socket(websocket: WebSocket) -> None: await _handle_question(state, message["text"]) elif msg_type == "passive": await _handle_passive(state, bool(message.get("enabled"))) + elif msg_type == "ritual_start": + await _handle_ritual_start(state) + elif msg_type == "ritual_step": + await _handle_ritual_step(state, message) + elif msg_type == "judgment": + await _handle_judgment(state, message) except WebSocketDisconnect: pass finally: diff --git a/backend/tests/test_judgment.py b/backend/tests/test_judgment.py new file mode 100644 index 0000000..9feba14 --- /dev/null +++ b/backend/tests/test_judgment.py @@ -0,0 +1,363 @@ +import random + +import pytest + +from app.judgment import ( + FAVOR_CORRECT_BANISH, + FAVOR_CORRECT_CROSS_OVER, + FAVOR_CORRECT_TRUST, + FAVOR_CROSS_OVER_FAIL, + FAVOR_TEST, + FAVOR_WRONG_BANISH, + FAVOR_WRONG_TRUST, + RITUAL_BASE_SUCCESS_CHANCE, + RITUAL_MIN_SUCCESS_CHANCE, + STUCK_VOLATILITY_THRESHOLD, + apply_favor_bias, + clamp_favor, + generate_tell, + is_stuck_spirit, + judge_verdict, + roll_ritual_success, +) +from app.inventory import CORRECT_JUDGMENT_ESSENCE, CROSS_OVER_ESSENCE + + +def _traits(alignment=0.5, power=0.5, volatility=0.5, deceptiveness=0.5): + return { + "alignment": alignment, + "power": power, + "volatility": volatility, + "deceptiveness": deceptiveness, + } + + +# --- clamp_favor ------------------------------------------------------- + + +def test_clamp_favor_within_range_unchanged(): + assert clamp_favor(0.3) == 0.3 + + +def test_clamp_favor_clamps_above_max(): + assert clamp_favor(5.0) == 1.0 + + +def test_clamp_favor_clamps_below_min(): + assert clamp_favor(-5.0) == -1.0 + + +def test_clamp_favor_at_exact_bounds(): + assert clamp_favor(1.0) == 1.0 + assert clamp_favor(-1.0) == -1.0 + + +# --- roll_ritual_success ------------------------------------------------- + + +def test_ritual_success_always_true_when_rng_below_chance(): + rng = random.Random() + # An easy entity (low power/deceptiveness) sits at the base chance; + # forcing rng.random() to 0 always beats any positive chance. + monkey_rng = random.Random(0) + monkey_rng.random = lambda: 0.0 # type: ignore[method-assign] + assert roll_ritual_success(_traits(power=0.0, deceptiveness=0.0), monkey_rng) is True + + +def test_ritual_success_always_false_when_rng_at_one(): + monkey_rng = random.Random(0) + monkey_rng.random = lambda: 0.999999 # type: ignore[method-assign] + assert roll_ritual_success(_traits(power=0.0, deceptiveness=0.0), monkey_rng) is False + + +def test_ritual_harder_entity_has_lower_success_chance(): + # Same rng draw, easy vs. hard entity: the hard one should fail where + # the easy one succeeds, for a draw threaded between the two chances. + easy = _traits(power=0.0, deceptiveness=0.0) + hard = _traits(power=1.0, deceptiveness=1.0) + + fixed_draw = (RITUAL_BASE_SUCCESS_CHANCE + RITUAL_MIN_SUCCESS_CHANCE) / 2 + + rng_easy = random.Random(0) + rng_easy.random = lambda: fixed_draw # type: ignore[method-assign] + rng_hard = random.Random(0) + rng_hard.random = lambda: fixed_draw # type: ignore[method-assign] + + assert roll_ritual_success(easy, rng_easy) is True + assert roll_ritual_success(hard, rng_hard) is False + + +def test_ritual_success_chance_never_below_floor(): + # Even the worst-case entity must be beatable — a low enough draw always + # succeeds. + rng = random.Random(0) + rng.random = lambda: RITUAL_MIN_SUCCESS_CHANCE - 0.01 # type: ignore[method-assign] + assert roll_ritual_success(_traits(power=1.0, deceptiveness=1.0), rng) is True + + +def test_ritual_success_alignment_and_volatility_dont_affect_odds(): + fixed_draw = 0.5 + rng_a = random.Random(0) + rng_a.random = lambda: fixed_draw # type: ignore[method-assign] + rng_b = random.Random(0) + rng_b.random = lambda: fixed_draw # type: ignore[method-assign] + + result_a = roll_ritual_success(_traits(alignment=0.0, volatility=0.0, power=0.4, deceptiveness=0.4), rng_a) + result_b = roll_ritual_success(_traits(alignment=1.0, volatility=1.0, power=0.4, deceptiveness=0.4), rng_b) + assert result_a == result_b + + +def test_ritual_success_uses_fresh_rng_by_default_and_varies(): + # No injected rng: repeated calls against the same traits should not + # all agree (proves it isn't signature/deterministically seeded). + outcomes = {roll_ritual_success(_traits()) for _ in range(200)} + assert outcomes == {True, False} + + +# --- is_stuck_spirit ----------------------------------------------------- + + +def test_stuck_spirit_requires_benevolent_and_high_volatility(): + assert is_stuck_spirit(_traits(alignment=0.7, volatility=0.9)) is True + + +def test_stuck_spirit_false_for_demon_even_if_volatile(): + assert is_stuck_spirit(_traits(alignment=0.2, volatility=0.95)) is False + + +def test_stuck_spirit_false_for_calm_benevolent_spirit(): + assert is_stuck_spirit(_traits(alignment=0.8, volatility=0.3)) is False + + +def test_stuck_spirit_boundary_volatility_not_stuck(): + assert is_stuck_spirit(_traits(alignment=0.9, volatility=STUCK_VOLATILITY_THRESHOLD)) is False + + +def test_stuck_spirit_boundary_alignment_is_stuck(): + assert is_stuck_spirit(_traits(alignment=0.5, volatility=0.99)) is True + + +# --- judge_verdict: trust ------------------------------------------------- + + +def test_trust_correct_on_benevolent_spirit(): + outcome = judge_verdict("trust", _traits(alignment=0.8)) + assert outcome.correct is True + assert outcome.consequence == "reward" + assert outcome.favor_delta == FAVOR_CORRECT_TRUST + assert outcome.essence_delta == CORRECT_JUDGMENT_ESSENCE + assert outcome.at_peace is False + + +def test_trust_wrong_on_demon(): + outcome = judge_verdict("trust", _traits(alignment=0.1)) + assert outcome.correct is False + assert outcome.consequence == "escalation" + assert outcome.favor_delta == FAVOR_WRONG_TRUST + assert outcome.essence_delta == 0 + + +def test_trust_boundary_alignment_counts_as_benevolent(): + outcome = judge_verdict("trust", _traits(alignment=0.5)) + assert outcome.correct is True + assert outcome.consequence == "reward" + + +# --- judge_verdict: banish ------------------------------------------------- + + +def test_banish_correct_on_demon(): + outcome = judge_verdict("banish", _traits(alignment=0.1)) + assert outcome.correct is True + assert outcome.consequence == "reward" + assert outcome.favor_delta == FAVOR_CORRECT_BANISH + assert outcome.essence_delta == CORRECT_JUDGMENT_ESSENCE + + +def test_banish_wrong_on_real_spirit(): + outcome = judge_verdict("banish", _traits(alignment=0.9)) + assert outcome.correct is False + assert outcome.consequence == "withdrawal" + assert outcome.favor_delta == FAVOR_WRONG_BANISH + assert outcome.essence_delta == 0 + + +def test_wrong_trust_penalty_larger_magnitude_than_wrong_banish(): + trust_outcome = judge_verdict("trust", _traits(alignment=0.0)) + banish_outcome = judge_verdict("banish", _traits(alignment=1.0)) + assert abs(trust_outcome.favor_delta) > abs(banish_outcome.favor_delta) + assert trust_outcome.favor_delta < 0 + assert banish_outcome.favor_delta < 0 + + +def test_all_favor_deltas_are_small_and_in_range(): + for delta in ( + FAVOR_CORRECT_TRUST, + FAVOR_CORRECT_BANISH, + FAVOR_WRONG_TRUST, + FAVOR_WRONG_BANISH, + FAVOR_CORRECT_CROSS_OVER, + FAVOR_CROSS_OVER_FAIL, + FAVOR_TEST, + ): + assert -1.0 <= delta <= 1.0 + assert abs(delta) <= 0.2 # single-digit percent of the [-1, 1] range + + +# --- judge_verdict: cross_over --------------------------------------------- + + +def test_cross_over_correct_on_stuck_spirit(): + outcome = judge_verdict("cross_over", _traits(alignment=0.8, volatility=0.9)) + assert outcome.correct is True + assert outcome.consequence == "crossed_over" + assert outcome.at_peace is True + assert outcome.favor_delta == FAVOR_CORRECT_CROSS_OVER + assert outcome.essence_delta == CROSS_OVER_ESSENCE + + +def test_cross_over_resisted_on_demon(): + outcome = judge_verdict("cross_over", _traits(alignment=0.1, volatility=0.9)) + assert outcome.correct is False + assert outcome.consequence == "resisted" + assert outcome.at_peace is False + assert outcome.favor_delta == 0 + assert outcome.essence_delta == 0 + + +def test_cross_over_resisted_on_non_stuck_real_spirit(): + outcome = judge_verdict("cross_over", _traits(alignment=0.8, volatility=0.2)) + assert outcome.correct is False + assert outcome.consequence == "resisted" + assert outcome.at_peace is False + assert outcome.favor_delta == 0 + assert outcome.essence_delta == 0 + + +def test_cross_over_essence_is_largest_reward_of_any_outcome(): + trust = judge_verdict("trust", _traits(alignment=0.9)) + banish = judge_verdict("banish", _traits(alignment=0.1)) + crossed = judge_verdict("cross_over", _traits(alignment=0.8, volatility=0.9)) + assert crossed.essence_delta > trust.essence_delta + assert crossed.essence_delta > banish.essence_delta + + +# --- judge_verdict: test --------------------------------------------------- + + +def test_test_verdict_without_completed_ritual_has_no_effect(): + outcome = judge_verdict("test", _traits(alignment=0.9), ritual_completed=False) + assert outcome.correct is False + assert outcome.consequence == "neutral" + assert outcome.favor_delta == 0 + assert outcome.essence_delta == 0 + assert outcome.at_peace is False + + +def test_test_verdict_with_successful_completed_ritual(): + outcome = judge_verdict( + "test", _traits(alignment=0.9), ritual_completed=True, ritual_success=True + ) + assert outcome.correct is True + assert outcome.consequence == "neutral" + assert outcome.favor_delta == 0 + assert outcome.essence_delta == 0 + + +def test_test_verdict_with_failed_completed_ritual(): + outcome = judge_verdict( + "test", _traits(alignment=0.9), ritual_completed=True, ritual_success=False + ) + assert outcome.correct is False + assert outcome.consequence == "neutral" + assert outcome.favor_delta == 0 + assert outcome.essence_delta == 0 + + +def test_judge_verdict_rejects_unknown_verdict(): + with pytest.raises(ValueError): + judge_verdict("smite", _traits()) + + +# --- apply_favor_bias ------------------------------------------------------- + + +def test_favor_bias_zero_favor_is_a_no_op(): + traits = _traits(volatility=0.5, deceptiveness=0.5) + biased = apply_favor_bias(traits, 0.0) + assert biased["volatility"] == pytest.approx(traits["volatility"]) + assert biased["deceptiveness"] == pytest.approx(traits["deceptiveness"]) + + +def test_favor_bias_positive_favor_lowers_volatility_and_deceptiveness(): + traits = _traits(volatility=0.5, deceptiveness=0.5) + biased = apply_favor_bias(traits, 1.0) + assert biased["volatility"] < traits["volatility"] + assert biased["deceptiveness"] < traits["deceptiveness"] + + +def test_favor_bias_negative_favor_raises_volatility_and_deceptiveness(): + traits = _traits(volatility=0.5, deceptiveness=0.5) + biased = apply_favor_bias(traits, -1.0) + assert biased["volatility"] > traits["volatility"] + assert biased["deceptiveness"] > traits["deceptiveness"] + + +def test_favor_bias_leaves_alignment_and_power_untouched(): + traits = _traits(alignment=0.3, power=0.7, volatility=0.5, deceptiveness=0.5) + biased = apply_favor_bias(traits, 1.0) + assert biased["alignment"] == traits["alignment"] + assert biased["power"] == traits["power"] + + +def test_favor_bias_clamps_to_valid_range(): + traits = _traits(volatility=0.02, deceptiveness=0.98) + biased = apply_favor_bias(traits, -1.0) + assert 0.0 <= biased["volatility"] <= 1.0 + assert 0.0 <= biased["deceptiveness"] <= 1.0 + biased_up = apply_favor_bias(traits, 1.0) + assert 0.0 <= biased_up["deceptiveness"] <= 1.0 + + +def test_favor_bias_effect_size_is_small(): + traits = _traits(volatility=0.5, deceptiveness=0.5) + biased = apply_favor_bias(traits, 1.0) + assert abs(biased["volatility"] - traits["volatility"]) <= 0.15 + assert abs(biased["deceptiveness"] - traits["deceptiveness"]) <= 0.15 + + +def test_favor_bias_out_of_range_favor_gets_clamped_first(): + traits = _traits(volatility=0.5, deceptiveness=0.5) + extreme = apply_favor_bias(traits, 5.0) + clamped = apply_favor_bias(traits, 1.0) + assert extreme == clamped + + +# --- generate_tell ----------------------------------------------------------- + + +def test_generate_tell_never_leaks_a_raw_number(): + rng = random.Random(42) + traits = _traits(alignment=0.9, power=0.1, volatility=0.95, deceptiveness=0.05) + for _ in range(50): + text = generate_tell(traits, rng) + assert isinstance(text, str) and text + # No digits anywhere in the line — the whole point is never + # surfacing a stat number. + assert not any(ch.isdigit() for ch in text) + + +def test_generate_tell_is_deterministic_given_same_rng_state(): + traits = _traits() + rng_a = random.Random(7) + rng_b = random.Random(7) + lines_a = [generate_tell(traits, rng_a) for _ in range(10)] + lines_b = [generate_tell(traits, rng_b) for _ in range(10)] + assert lines_a == lines_b + + +def test_generate_tell_varies_across_draws(): + traits = _traits() + rng = random.Random(99) + lines = {generate_tell(traits, rng) for _ in range(30)} + assert len(lines) > 1 diff --git a/backend/tests/test_ws_ritual_judgment.py b/backend/tests/test_ws_ritual_judgment.py new file mode 100644 index 0000000..5f8cbce --- /dev/null +++ b/backend/tests/test_ws_ritual_judgment.py @@ -0,0 +1,614 @@ +"""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"): + 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)) + + +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"]