Merge Workstream C: unlocks + inventory items + sigils + essence

backend/app/inventory.py

@@ -0,0 +1,219 @@
+"""Unlocks, inventory items, sigils, and the essence economy — Workstream C
+of docs/superpowers/specs/2026-07-23-character-depth-ghost-log-design.md.
+
+Essence amounts (the implementer's call per the spec's "exact amounts are
+the implementer's call; keep summon income small (single digits) and
+milestone income larger (tens)" guidance):
+  - SUMMON_ESSENCE_TRICKLE: every successful summon, any mode/rarity — kept
+    tiny so it accrues constantly without needing any gating logic. This is
+    the only trigger point that exists in `backend/app/ws.py` today (see
+    module docstring there); it's wired directly in `_handle_summon`.
+  - RITUAL_SUCCESS_ESSENCE / CORRECT_JUDGMENT_ESSENCE / CROSS_OVER_ESSENCE:
+    milestone constants for Workstream B's ritual/judgment WS handlers to
+    draw from once they land (those handlers — `ritual_start`,
+    `ritual_step`, `judgment` — don't exist in this worktree yet; Workstream
+    B owns `backend/app/judgment.py` and the WS wiring per the spec). They're
+    defined here, alongside the drop table, so both workstreams price things
+    out of one shared module instead of duplicating numbers. Cross-over pays
+    the most, matching the contract's "the largest essence reward of any
+    outcome".
+
+Item drop odds (also the implementer's call): ritual success and correct
+judgments are deliberate, effortful player actions, so they roll a bit more
+generously than a bare summon (which happens constantly and is often
+passive/ambient). High-rarity summons scale with how rare the entity itself
+already is — a mythic summon is already a jackpot, so the drop on top of it
+pays off further. None of these odds are large enough to make items feel
+guaranteed/grindy, matching the same "reachable within a handful of
+sessions, not grindy" spirit as the essence guidance.
+"""
+
+import random
+import uuid
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.unlock import UnlockRecord
+from app.models.user import User
+
+# --- Essence economy --------------------------------------------------
+
+SUMMON_ESSENCE_TRICKLE = 2  # every successful summon, any mode
+RITUAL_SUCCESS_ESSENCE = 15  # a completed ritual
+CORRECT_JUDGMENT_ESSENCE = 12  # correct trust/banish call
+CROSS_OVER_ESSENCE = 25  # correct cross_over — largest reward of any outcome
+
+# --- Unlocks ------------------------------------------------------------
+
+UNLOCK_PRICES: dict[str, int] = {
+    "listening_tool": 40,
+}
+
+# --- Item drops -----------------------------------------------------------
+
+HIGH_RARITY_TIERS = {"rare", "mythic"}
+
+# (trigger key) -> drop chance. Triggers correspond to the contract's
+# `item_drop` emission points: "after a correct judgment, a successful
+# ritual, or a high-rarity summon" — the last one is split by tier since a
+# mythic summon should feel more rewarded than a merely-rare one.
+DROP_CHANCES: dict[str, float] = {
+    "judgment": 0.20,
+    "ritual": 0.25,
+    "summon_rare": 0.15,
+    "summon_mythic": 0.35,
+}
+
+# (trigger key) -> (item_type, item_key pool)
+ITEM_POOLS: dict[str, tuple[str, list[str]]] = {
+    "judgment": (
+        "trinket",
+        ["static_shard", "cracked_locket", "cold_coin", "grave_dust"],
+    ),
+    "ritual": (
+        "relic",
+        ["obsidian_mirror", "bone_dial", "silver_tuning_fork", "warded_chalk"],
+    ),
+    "summon_rare": (
+        "curio",
+        ["moth_wing", "ectoplasm_vial", "tarnished_key"],
+    ),
+    "summon_mythic": (
+        "curio",
+        ["black_candle_stub", "veil_thread", "gilded_grave_dust"],
+    ),
+}
+
+
+def summon_drop_trigger(rarity: str) -> str | None:
+    """Maps an entity's rarity tier to its drop-trigger key, or None if the
+    summon isn't high-rarity enough to qualify for a drop roll at all."""
+    if rarity == "rare":
+        return "summon_rare"
+    if rarity == "mythic":
+        return "summon_mythic"
+    return None
+
+
+def roll_item_drop(trigger: str, rng: random.Random | None = None) -> dict | None:
+    """Rolls for an item drop at one of the contract's trigger points
+    ("judgment", "ritual", "summon_rare", "summon_mythic"). Returns the
+    `item_drop` frame's `item` payload shape
+    (`{"item_type", "item_key", "payload"}`), or None on a miss / unknown
+    trigger."""
+    rng = rng if rng is not None else random.Random()
+    chance = DROP_CHANCES.get(trigger)
+    pool = ITEM_POOLS.get(trigger)
+    if chance is None or pool is None:
+        return None
+    if rng.random() >= chance:
+        return None
+    item_type, keys = pool
+    return {
+        "item_type": item_type,
+        "item_key": rng.choice(keys),
+        "payload": {"trigger": trigger},
+    }
+
+
+def credit_essence(user: User, amount: int) -> int:
+    """Adjusts `user.essence` in place, floored at 0, and returns the new
+    balance. Caller is responsible for committing/flushing."""
+    user.essence = max(0, user.essence + amount)
+    return user.essence
+
+
+# --- Sigil design validation ---------------------------------------------
+
+SIGIL_MAX_POINTS = 12
+_RUNE_MAX_LEN = 32
+
+
+def validate_sigil_design(design: object) -> dict | None:
+    """Validates the placeholder sigil design shape
+    `{"points": [[x, y], ...], "rune": str}` (a stand-in until Workstream F's
+    sigil designer defines the real shape — see the spec's Workstream C
+    section). `points` is capped at `SIGIL_MAX_POINTS` to bound payload
+    size. Returns a normalized dict on success, or None if the shape is
+    invalid."""
+    if not isinstance(design, dict):
+        return None
+
+    points = design.get("points")
+    if not isinstance(points, list) or not (1 <= len(points) <= SIGIL_MAX_POINTS):
+        return None
+
+    normalized_points: list[list[float]] = []
+    for point in points:
+        if (
+            not isinstance(point, (list, tuple))
+            or len(point) != 2
+            or not all(
+                isinstance(coord, (int, float)) and not isinstance(coord, bool)
+                for coord in point
+            )
+        ):
+            return None
+        normalized_points.append([float(point[0]), float(point[1])])
+
+    rune = design.get("rune")
+    if not isinstance(rune, str) or not (1 <= len(rune) <= _RUNE_MAX_LEN):
+        return None
+
+    return {"points": normalized_points, "rune": rune}
+
+
+# --- Purchases --------------------------------------------------------
+
+
+class InsufficientEssenceError(Exception):
+    """Raised when a user tries to buy an unlock they can't afford."""
+
+
+class UnknownUnlockError(Exception):
+    """Raised when `unlock_key` isn't in `UNLOCK_PRICES`."""
+
+
+async def purchase_unlock(
+    db: AsyncSession, user_id: uuid.UUID, unlock_key: str
+) -> UnlockRecord:
+    """Atomically spends essence for `unlock_key` and records the unlock.
+
+    Locks the user's row (`SELECT ... FOR UPDATE`) for the duration of the
+    balance check + deduction, so two concurrent purchase requests can't
+    both read the same starting balance and both succeed (double-spend).
+    The second request blocks on the row lock until the first commits, then
+    re-reads the now-decremented balance. Idempotent: re-buying an unlock
+    already owned returns the existing record without charging again.
+    """
+    if unlock_key not in UNLOCK_PRICES:
+        raise UnknownUnlockError(unlock_key)
+    price = UNLOCK_PRICES[unlock_key]
+
+    locked_user = await db.scalar(
+        select(User).where(User.id == user_id).with_for_update()
+    )
+    if locked_user is None:
+        raise UnknownUnlockError(unlock_key)  # user vanished mid-request
+
+    existing = await db.scalar(
+        select(UnlockRecord).where(
+            UnlockRecord.user_id == user_id, UnlockRecord.unlock_key == unlock_key
+        )
+    )
+    if existing is not None:
+        return existing
+
+    if locked_user.essence < price:
+        raise InsufficientEssenceError(
+            f"not enough essence for {unlock_key!r}: "
+            f"have {locked_user.essence}, need {price}"
+        )
+
+    locked_user.essence -= price
+    record = UnlockRecord(user_id=user_id, unlock_key=unlock_key)
+    db.add(record)
+    await db.commit()
+    await db.refresh(record)
+    return record

backend/app/main.py

@@ -13,6 +13,7 @@ from app.config import settings
 from app.db import Base, async_session_maker, engine
 from app.routes.auth import router as auth_router
 from app.routes.codex import router as codex_router
+from app.routes.inventory import router as inventory_router
 from app.routes.shop import router as shop_router
 from app.session_cleanup import delete_expired_sessions
 from app.ws import AUDIO_DIR
@@ -64,6 +65,7 @@ async def lifespan(app: FastAPI):
 app = FastAPI(title="Quantumancy", lifespan=lifespan)
 app.include_router(auth_router)
 app.include_router(codex_router)
+app.include_router(inventory_router)
 app.include_router(shop_router)
 app.include_router(ws_router)
 

backend/app/models/__init__.py

@@ -3,6 +3,9 @@ from app.models.contact_session import ContactSession
 from app.models.entity import Entity
 from app.models.entity_sighting import EntitySighting
 from app.models.event import Event
+from app.models.inventory_item import InventoryItem
+from app.models.sigil import Sigil
+from app.models.unlock import UnlockRecord
 from app.models.user import User
 from app.models.waitlist_entry import WaitlistEntry
 
@@ -13,5 +16,8 @@ __all__ = [
     "Entity",
     "EntitySighting",
     "Event",
+    "InventoryItem",
+    "Sigil",
+    "UnlockRecord",
     "WaitlistEntry",
 ]

backend/app/models/inventory_item.py

@@ -0,0 +1,24 @@
+import uuid
+from datetime import datetime, timezone
+
+from sqlalchemy import DateTime, ForeignKey, String
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db import Base
+
+
+class InventoryItem(Base):
+    """An item a seeker has been dropped (see `app.inventory.roll_item_drop`)
+    — a Reliquary trinket, not a purchased unlock."""
+
+    __tablename__ = "inventory_items"
+
+    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
+    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"), index=True)
+    item_type: Mapped[str] = mapped_column(String(32), index=True)
+    item_key: Mapped[str] = mapped_column(String(64))
+    payload: Mapped[dict] = mapped_column(JSONB, default=dict)
+    obtained_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )

backend/app/models/sigil.py

@@ -0,0 +1,25 @@
+import uuid
+from datetime import datetime, timezone
+
+from sqlalchemy import DateTime, ForeignKey, String
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db import Base
+
+
+class Sigil(Base):
+    """A seeker-designed sigil (Workstream F's constrained geometric
+    builder). `design` is placeholder-shaped `{"points": [[x,y],...],
+    "rune": str}` pending the frontend sigil-designer workstream landing —
+    see `app.inventory.validate_sigil_design`."""
+
+    __tablename__ = "sigils"
+
+    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
+    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"), index=True)
+    name: Mapped[str] = mapped_column(String(64))
+    design: Mapped[dict] = mapped_column(JSONB, default=dict)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )

backend/app/models/unlock.py

@@ -0,0 +1,21 @@
+import uuid
+from datetime import datetime, timezone
+
+from sqlalchemy import DateTime, ForeignKey, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db import Base
+
+
+class UnlockRecord(Base):
+    """A permanent unlock a seeker has purchased with essence (e.g. the
+    listening tool). One row per (user, unlock_key)."""
+
+    __tablename__ = "unlocks"
+
+    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
+    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"), index=True)
+    unlock_key: Mapped[str] = mapped_column(String(64), index=True)
+    unlocked_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )

backend/app/models/user.py

@@ -1,7 +1,7 @@
 import uuid
 from datetime import datetime, timezone
 
-from sqlalchemy import DateTime, String
+from sqlalchemy import DateTime, Integer, String
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db import Base
@@ -14,6 +14,13 @@ 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)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
     )

backend/app/routes/auth.py

@@ -7,6 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.db import get_db
 from app.deps import SESSION_COOKIE_NAME, get_current_user
 from app.models.auth_session import AuthSession, SESSION_TTL, generate_session_token, hash_token
+from app.models.unlock import UnlockRecord
 from app.models.user import User
 from app.schemas import LoginRequest, RegisterRequest, UserOut
 from app.security import hash_password, verify_password
@@ -92,5 +93,13 @@ async def logout(
 
 
 @router.get("/me", response_model=UserOut)
-async def me(user: User = Depends(get_current_user)):
+async def me(
-    return user
+    user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)
+):
+    result = await db.execute(
+        select(UnlockRecord.unlock_key).where(UnlockRecord.user_id == user.id)
+    )
+    unlock_keys = [row[0] for row in result.all()]
+    return UserOut(
+        id=user.id, username=user.username, essence=user.essence, unlocks=unlock_keys
+    )

backend/app/routes/inventory.py

@@ -0,0 +1,119 @@
+"""The Reliquary: a seeker's unlocks, dropped items, and saved sigils —
+Workstream C of docs/superpowers/specs/2026-07-23-character-depth-ghost-log-design.md."""
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db import get_db
+from app.deps import get_current_user
+from app.inventory import (
+    UNLOCK_PRICES,
+    InsufficientEssenceError,
+    UnknownUnlockError,
+    purchase_unlock,
+    validate_sigil_design,
+)
+from app.models.inventory_item import InventoryItem
+from app.models.sigil import Sigil
+from app.models.unlock import UnlockRecord
+from app.models.user import User
+from app.rate_limit import RateLimiter
+from app.schemas import (
+    InventoryItemOut,
+    PurchaseOut,
+    SigilIn,
+    SigilOut,
+    UnlockOut,
+)
+
+router = APIRouter(prefix="/api/inventory", tags=["inventory"])
+
+# A seeker mashing the buy button shouldn't be able to spam the DB — the
+# essence balance check itself is race-safe (see app.inventory.purchase_unlock)
+# but there's no reason to let unlimited attempts through either.
+purchase_limiter = RateLimiter(max_requests=20, window_seconds=60)
+
+
+@router.get("/unlocks", response_model=list[UnlockOut])
+async def list_unlocks(
+    user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)
+):
+    result = await db.execute(
+        select(UnlockRecord)
+        .where(UnlockRecord.user_id == user.id)
+        .order_by(UnlockRecord.unlocked_at)
+    )
+    return result.scalars().all()
+
+
+@router.get("/items", response_model=list[InventoryItemOut])
+async def list_items(
+    user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)
+):
+    result = await db.execute(
+        select(InventoryItem)
+        .where(InventoryItem.user_id == user.id)
+        .order_by(InventoryItem.obtained_at)
+    )
+    return result.scalars().all()
+
+
+@router.get("/sigils", response_model=list[SigilOut])
+async def list_sigils(
+    user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)
+):
+    result = await db.execute(
+        select(Sigil).where(Sigil.user_id == user.id).order_by(Sigil.created_at)
+    )
+    return result.scalars().all()
+
+
+@router.post("/sigils", response_model=SigilOut, status_code=status.HTTP_201_CREATED)
+async def save_sigil(
+    payload: SigilIn,
+    user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    normalized = validate_sigil_design(payload.design)
+    if normalized is None:
+        raise HTTPException(
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+            "sigil design must be {\"points\": [[x, y], ...] (1-12 points), "
+            "\"rune\": str}",
+        )
+
+    sigil = Sigil(user_id=user.id, name=payload.name, design=normalized)
+    db.add(sigil)
+    await db.commit()
+    await db.refresh(sigil)
+    return sigil
+
+
+@router.post("/unlocks/{unlock_key}", response_model=PurchaseOut)
+async def buy_unlock(
+    unlock_key: str,
+    user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    if unlock_key not in UNLOCK_PRICES:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "no such unlock")
+
+    if not purchase_limiter.allow(str(user.id)):
+        raise HTTPException(
+            status.HTTP_429_TOO_MANY_REQUESTS, "too many purchase attempts — slow down"
+        )
+
+    try:
+        record = await purchase_unlock(db, user.id, unlock_key)
+    except InsufficientEssenceError as exc:
+        raise HTTPException(status.HTTP_402_PAYMENT_REQUIRED, str(exc)) from exc
+    except UnknownUnlockError as exc:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "no such unlock") from exc
+
+    await db.refresh(user)
+    return PurchaseOut(
+        unlock_key=record.unlock_key,
+        unlocked_at=record.unlocked_at,
+        essence=user.essence,
+    )

backend/app/schemas.py

@@ -1,4 +1,5 @@
 import uuid
+from datetime import datetime
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -12,6 +13,8 @@ class RegisterRequest(BaseModel):
 class UserOut(BaseModel):
     id: uuid.UUID
     username: str
+    essence: int = 0
+    unlocks: list[str] = Field(default_factory=list)
 
     model_config = ConfigDict(from_attributes=True)
 
@@ -19,3 +22,40 @@ class UserOut(BaseModel):
 class LoginRequest(BaseModel):
     username: str
     password: str
+
+
+class UnlockOut(BaseModel):
+    unlock_key: str
+    unlocked_at: datetime
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class PurchaseOut(BaseModel):
+    unlock_key: str
+    unlocked_at: datetime
+    essence: int
+
+
+class InventoryItemOut(BaseModel):
+    id: uuid.UUID
+    item_type: str
+    item_key: str
+    payload: dict
+    obtained_at: datetime
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class SigilIn(BaseModel):
+    name: str = Field(min_length=1, max_length=64)
+    design: dict
+
+
+class SigilOut(BaseModel):
+    id: uuid.UUID
+    name: str
+    design: dict
+    created_at: datetime
+
+    model_config = ConfigDict(from_attributes=True)

backend/app/ws.py

@@ -30,12 +30,15 @@ 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.llm.service import SpiritBusyError, spirit_service
 from app.models.auth_session import AuthSession, hash_token
 from app.models.contact_session import ContactSession
 from app.models.entity import Entity
 from app.models.entity_sighting import EntitySighting
 from app.models.event import Event
+from app.models.inventory_item import InventoryItem
+from app.models.user import User
 from app.possession import compute_stability
 from app.rate_limit import RateLimiter, resolve_client_ip
 from app.telemetry import detect_wire_spike, sample_network
@@ -247,6 +250,45 @@ async def _summon(state: SeanceState, channel: str) -> tuple[Entity, bool]:
         return entity, is_new
 
 
+async def _reward_summon(state: SeanceState) -> None:
+    """Workstream C's essence-trickle + item-drop trigger points that exist
+    in this handler today: a small essence trickle for every successful
+    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."""
+    assert state.entity is not None
+    rarity = state.entity.get("rarity", "common")
+
+    async with session_maker() as db:
+        user = await db.get(User, state.user_id)
+        if user is None:
+            return
+        credit_essence(user, SUMMON_ESSENCE_TRICKLE)
+
+        item = None
+        trigger = summon_drop_trigger(rarity)
+        if trigger is not None:
+            item = roll_item_drop(trigger)
+            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_summon(state: SeanceState) -> None:
     # Short-circuits: an account already over its own cap never gets far
     # enough to spend from the IP budget too.
@@ -269,6 +311,7 @@ async def _handle_summon(state: SeanceState) -> None:
     await state.send_queue.put(
         {"type": "entity", "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)
 

backend/tests/test_inventory.py

@@ -0,0 +1,352 @@
+import random
+import uuid
+
+import pytest
+from sqlalchemy import select
+
+from app.inventory import (
+    DROP_CHANCES,
+    HIGH_RARITY_TIERS,
+    ITEM_POOLS,
+    SIGIL_MAX_POINTS,
+    UNLOCK_PRICES,
+    InsufficientEssenceError,
+    UnknownUnlockError,
+    credit_essence,
+    purchase_unlock,
+    roll_item_drop,
+    summon_drop_trigger,
+    validate_sigil_design,
+)
+from app.models.unlock import UnlockRecord
+from app.models.user import User
+
+# ---------------------------------------------------------------------------
+# Drop-roll probability sanity — seeded RNG, so this is a deterministic
+# statistical check, not a flaky one.
+# ---------------------------------------------------------------------------
+
+
+def test_roll_item_drop_statistical_sanity_summon_mythic():
+    rng = random.Random(1234)
+    trials = 20_000
+    hits = sum(1 for _ in range(trials) if roll_item_drop("summon_mythic", rng) is not None)
+    empirical = hits / trials
+    expected = DROP_CHANCES["summon_mythic"]
+    assert abs(empirical - expected) < 0.02
+
+
+def test_roll_item_drop_statistical_sanity_judgment():
+    rng = random.Random(5678)
+    trials = 20_000
+    hits = sum(1 for _ in range(trials) if roll_item_drop("judgment", rng) is not None)
+    empirical = hits / trials
+    expected = DROP_CHANCES["judgment"]
+    assert abs(empirical - expected) < 0.02
+
+
+def test_roll_item_drop_is_deterministic_given_same_seed():
+    results_a = [roll_item_drop("ritual", random.Random(42)) for _ in range(50)]
+    results_b = [roll_item_drop("ritual", random.Random(42)) for _ in range(50)]
+    assert results_a == results_b
+
+
+def test_roll_item_drop_unknown_trigger_returns_none():
+    assert roll_item_drop("not_a_real_trigger", random.Random(1)) is None
+
+
+def test_roll_item_drop_shape_matches_item_pool():
+    # A high roll chance with a fixed seed known to hit, to check the shape.
+    rng = random.Random(1234)
+    item = None
+    for _ in range(200):
+        item = roll_item_drop("ritual", rng)
+        if item is not None:
+            break
+    assert item is not None
+    item_type, keys = ITEM_POOLS["ritual"]
+    assert item["item_type"] == item_type
+    assert item["item_key"] in keys
+    assert item["payload"] == {"trigger": "ritual"}
+
+
+def test_summon_drop_trigger_only_fires_for_high_rarity():
+    assert summon_drop_trigger("common") is None
+    assert summon_drop_trigger("uncommon") is None
+    assert summon_drop_trigger("rare") == "summon_rare"
+    assert summon_drop_trigger("mythic") == "summon_mythic"
+    assert HIGH_RARITY_TIERS == {"rare", "mythic"}
+
+
+# ---------------------------------------------------------------------------
+# credit_essence
+# ---------------------------------------------------------------------------
+
+
+def test_credit_essence_adds_and_floors_at_zero():
+    user = User(username="x", password_hash="x")
+    user.essence = 5
+    assert credit_essence(user, 3) == 8
+    assert credit_essence(user, -100) == 0
+
+
+# ---------------------------------------------------------------------------
+# Sigil design validation
+# ---------------------------------------------------------------------------
+
+
+def test_validate_sigil_design_accepts_valid_shape():
+    design = {"points": [[0, 0], [1.5, 2.5], [3, -3]], "rune": "eld"}
+    normalized = validate_sigil_design(design)
+    assert normalized == {"points": [[0.0, 0.0], [1.5, 2.5], [3.0, -3.0]], "rune": "eld"}
+
+
+def test_validate_sigil_design_rejects_too_many_points():
+    design = {"points": [[i, i] for i in range(SIGIL_MAX_POINTS + 1)], "rune": "eld"}
+    assert validate_sigil_design(design) is None
+
+
+def test_validate_sigil_design_accepts_max_points():
+    design = {"points": [[i, i] for i in range(SIGIL_MAX_POINTS)], "rune": "eld"}
+    assert validate_sigil_design(design) is not None
+
+
+def test_validate_sigil_design_rejects_empty_points():
+    assert validate_sigil_design({"points": [], "rune": "eld"}) is None
+
+
+def test_validate_sigil_design_rejects_bad_point_shape():
+    assert validate_sigil_design({"points": [[1, 2, 3]], "rune": "eld"}) is None
+    assert validate_sigil_design({"points": [["a", "b"]], "rune": "eld"}) is None
+    assert validate_sigil_design({"points": "not-a-list", "rune": "eld"}) is None
+
+
+def test_validate_sigil_design_rejects_missing_or_bad_rune():
+    assert validate_sigil_design({"points": [[0, 0]]}) is None
+    assert validate_sigil_design({"points": [[0, 0]], "rune": ""}) is None
+    assert validate_sigil_design({"points": [[0, 0]], "rune": 5}) is None
+    assert validate_sigil_design({"points": [[0, 0]], "rune": "x" * 33}) is None
+
+
+def test_validate_sigil_design_rejects_non_dict():
+    assert validate_sigil_design("nope") is None
+    assert validate_sigil_design(None) is None
+
+
+# ---------------------------------------------------------------------------
+# purchase_unlock — success, insufficient funds, unknown unlock, idempotency,
+# race safety.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_purchase_unlock_success_deducts_essence_and_records(db_session):
+    user = User(username="buyer", password_hash="x", essence=100)
+    db_session.add(user)
+    await db_session.commit()
+    await db_session.refresh(user)
+
+    price = UNLOCK_PRICES["listening_tool"]
+    record = await purchase_unlock(db_session, user.id, "listening_tool")
+    assert record.unlock_key == "listening_tool"
+
+    await db_session.refresh(user)
+    assert user.essence == 100 - price
+
+    rows = (
+        await db_session.execute(
+            select(UnlockRecord).where(UnlockRecord.user_id == user.id)
+        )
+    ).scalars().all()
+    assert len(rows) == 1
+    assert rows[0].unlock_key == "listening_tool"
+
+
+@pytest.mark.asyncio
+async def test_purchase_unlock_insufficient_funds_raises_and_does_not_charge(db_session):
+    user = User(username="pauper", password_hash="x", essence=5)
+    db_session.add(user)
+    await db_session.commit()
+    await db_session.refresh(user)
+
+    with pytest.raises(InsufficientEssenceError):
+        await purchase_unlock(db_session, user.id, "listening_tool")
+
+    await db_session.refresh(user)
+    assert user.essence == 5
+    rows = (
+        await db_session.execute(
+            select(UnlockRecord).where(UnlockRecord.user_id == user.id)
+        )
+    ).scalars().all()
+    assert rows == []
+
+
+@pytest.mark.asyncio
+async def test_purchase_unlock_unknown_key_raises(db_session):
+    user = User(username="curious", password_hash="x", essence=1000)
+    db_session.add(user)
+    await db_session.commit()
+    await db_session.refresh(user)
+
+    with pytest.raises(UnknownUnlockError):
+        await purchase_unlock(db_session, user.id, "does_not_exist")
+
+
+@pytest.mark.asyncio
+async def test_purchase_unlock_is_idempotent_does_not_double_charge(db_session):
+    user = User(username="rebuyer", password_hash="x", essence=100)
+    db_session.add(user)
+    await db_session.commit()
+    await db_session.refresh(user)
+
+    price = UNLOCK_PRICES["listening_tool"]
+    await purchase_unlock(db_session, user.id, "listening_tool")
+    await db_session.refresh(user)
+    assert user.essence == 100 - price
+
+    # Buying again must not charge a second time.
+    await purchase_unlock(db_session, user.id, "listening_tool")
+    await db_session.refresh(user)
+    assert user.essence == 100 - price
+
+    rows = (
+        await db_session.execute(
+            select(UnlockRecord).where(UnlockRecord.user_id == user.id)
+        )
+    ).scalars().all()
+    assert len(rows) == 1
+
+
+# ---------------------------------------------------------------------------
+# HTTP layer: purchase endpoint, list endpoints, sigil save, /auth/me shape.
+# ---------------------------------------------------------------------------
+
+
+async def _register_and_login(client, username="seeker"):
+    await client.post(
+        "/auth/register", json={"username": username, "password": "spookyspooky"}
+    )
+    resp = await client.post(
+        "/auth/login", json={"username": username, "password": "spookyspooky"}
+    )
+    return resp.json()["id"]
+
+
+@pytest.mark.asyncio
+async def test_buy_unlock_success_over_http(client, db_session):
+    user_id = await _register_and_login(client, "affluent")
+    user = await db_session.get(User, uuid.UUID(user_id))
+    user.essence = 100
+    await db_session.commit()
+
+    resp = await client.post("/api/inventory/unlocks/listening_tool")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["unlock_key"] == "listening_tool"
+    assert body["essence"] == 100 - UNLOCK_PRICES["listening_tool"]
+
+    me = await client.get("/auth/me")
+    assert me.json()["unlocks"] == ["listening_tool"]
+    assert me.json()["essence"] == 100 - UNLOCK_PRICES["listening_tool"]
+
+
+@pytest.mark.asyncio
+async def test_buy_unlock_insufficient_funds_over_http(client, db_session):
+    await _register_and_login(client, "broke")
+    # A fresh user starts with 0 essence.
+    resp = await client.post("/api/inventory/unlocks/listening_tool")
+    assert resp.status_code == 402
+
+    me = await client.get("/auth/me")
+    assert me.json()["unlocks"] == []
+    assert me.json()["essence"] == 0
+
+
+@pytest.mark.asyncio
+async def test_buy_unlock_unknown_key_over_http(client):
+    await _register_and_login(client, "explorer")
+    resp = await client.post("/api/inventory/unlocks/nonexistent_thing")
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_buy_unlock_requires_auth(client):
+    resp = await client.post("/api/inventory/unlocks/listening_tool")
+    assert resp.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_inventory_list_endpoints_require_auth(client):
+    assert (await client.get("/api/inventory/unlocks")).status_code == 401
+    assert (await client.get("/api/inventory/items")).status_code == 401
+    assert (await client.get("/api/inventory/sigils")).status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_list_unlocks_and_items_empty_then_populated(client, db_session):
+    user_id = await _register_and_login(client, "collector")
+
+    empty = await client.get("/api/inventory/unlocks")
+    assert empty.status_code == 200
+    assert empty.json() == []
+
+    user = await db_session.get(User, uuid.UUID(user_id))
+    user.essence = 100
+    await db_session.commit()
+    await client.post("/api/inventory/unlocks/listening_tool")
+
+    populated = await client.get("/api/inventory/unlocks")
+    assert len(populated.json()) == 1
+    assert populated.json()[0]["unlock_key"] == "listening_tool"
+    assert "unlocked_at" in populated.json()[0]
+
+
+@pytest.mark.asyncio
+async def test_save_and_list_sigil(client):
+    await _register_and_login(client, "sigilsmith")
+
+    resp = await client.post(
+        "/api/inventory/sigils",
+        json={"name": "ward-of-quiet", "design": {"points": [[0, 0], [1, 1]], "rune": "eld"}},
+    )
+    assert resp.status_code == 201
+    body = resp.json()
+    assert body["name"] == "ward-of-quiet"
+    assert body["design"]["rune"] == "eld"
+
+    listed = await client.get("/api/inventory/sigils")
+    assert len(listed.json()) == 1
+    assert listed.json()[0]["name"] == "ward-of-quiet"
+
+
+@pytest.mark.asyncio
+async def test_save_sigil_rejects_too_many_points(client):
+    await _register_and_login(client, "greedy-sigilsmith")
+    resp = await client.post(
+        "/api/inventory/sigils",
+        json={
+            "name": "overreach",
+            "design": {"points": [[i, i] for i in range(13)], "rune": "eld"},
+        },
+    )
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_save_sigil_rejects_missing_rune(client):
+    await _register_and_login(client, "runeless")
+    resp = await client.post(
+        "/api/inventory/sigils",
+        json={"name": "blank", "design": {"points": [[0, 0]]}},
+    )
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_auth_me_includes_essence_and_unlocks_default(client):
+    await _register_and_login(client, "freshuser")
+    resp = await client.get("/auth/me")
+    body = resp.json()
+    assert body["essence"] == 0
+    assert body["unlocks"] == []

backend/tests/test_ws_session.py

@@ -1,12 +1,16 @@
 import asyncio
+import uuid
 
 import pytest
 from sqlalchemy import select
 
 import app.ws
 from app.entities import fallback_profile
+from app.inventory import SUMMON_ESSENCE_TRICKLE
 from app.models.contact_session import ContactSession
 from app.models.event import Event
+from app.models.inventory_item import InventoryItem
+from app.models.user import User
 from app.rate_limit import RateLimiter
 
 
@@ -266,3 +270,168 @@ async def test_summon_per_ip_bucket_follows_cf_connecting_ip_not_socket_peer(
         # A different visitor IP behind the same tunnel — must not be
         # rejected by visitor A's already-spent bucket.
         _read_until(ws, "entity")
+
+
+@pytest.mark.asyncio
+async def test_summon_credits_essence_trickle(sync_client, db_session, monkeypatch):
+    # The module-level summon_ip_limiter/summon_limiter are shared singletons
+    # that persist real hit counts across every test in this file (they're
+    # only overridden where a test explicitly monkeypatches them, as earlier
+    # tests above do) — fresh, generous instances here keep this test from
+    # depending on how many summons ran before it.
+    monkeypatch.setattr(app.ws, "summon_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60))
+    monkeypatch.setattr(app.ws, "summon_limiter", RateLimiter(max_requests=1000, window_seconds=60))
+
+    token = _login(sync_client, "trickle-earner")
+    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")
+        ws.send_json({"type": "summon"})
+        _read_until(ws, "entity")
+        _read_until(ws, "utterance", kind="greeting")
+
+    for _ in range(40):
+        db_session.expire_all()
+        user = await db_session.get(User, user_id)
+        if user.essence == SUMMON_ESSENCE_TRICKLE:
+            break
+        await asyncio.sleep(0.05)
+    assert user.essence == SUMMON_ESSENCE_TRICKLE
+
+
+@pytest.mark.asyncio
+async def test_summon_high_rarity_item_drop_is_persisted_and_sent(
+    sync_client, db_session, monkeypatch
+):
+    # Force a hit so the wiring is exercised deterministically rather than
+    # relying on the real roll odds (those are covered statistically in
+    # test_inventory.py).
+    forced_item = {
+        "item_type": "curio",
+        "item_key": "veil_thread",
+        "payload": {"trigger": "summon_mythic"},
+    }
+    monkeypatch.setattr(app.ws, "roll_item_drop", lambda trigger: forced_item)
+    # See test_summon_credits_essence_trickle above re: why this is reset.
+    monkeypatch.setattr(app.ws, "summon_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60))
+    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"):
+            profile = fallback_profile(signature)
+            profile["rarity"] = "mythic"
+            return profile
+
+        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():
+                yield "hi"
+
+            return gen()
+
+        def ambient_ready(self):
+            return False
+
+    monkeypatch.setattr(app.ws, "spirit_service", MythicSpiritService())
+
+    token = _login(sync_client, "jackpot-seeker")
+    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")
+        ws.send_json({"type": "summon"})
+        entity_frame = _read_until(ws, "entity")
+        assert entity_frame["entity"]["rarity"] == "mythic"
+        drop_frame = _read_until(ws, "item_drop")
+        assert drop_frame["item"] == forced_item
+
+    for _ in range(40):
+        db_session.expire_all()
+        items = (
+            await db_session.execute(
+                select(InventoryItem).where(InventoryItem.user_id == user_id)
+            )
+        ).scalars().all()
+        if items:
+            break
+        await asyncio.sleep(0.05)
+    assert len(items) == 1
+    assert items[0].item_key == "veil_thread"
+    assert items[0].item_type == "curio"
+
+
+@pytest.mark.asyncio
+async def test_summon_common_rarity_does_not_roll_a_drop(sync_client, db_session, monkeypatch):
+    # See test_summon_credits_essence_trickle above re: why this is reset.
+    monkeypatch.setattr(app.ws, "summon_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60))
+    monkeypatch.setattr(app.ws, "summon_limiter", RateLimiter(max_requests=1000, window_seconds=60))
+
+    # Common is the overwhelmingly likely fallback rarity (55% weight) but
+    # not guaranteed — force it explicitly so this test can't flake.
+    def _common_only(signature: str) -> dict:
+        profile = fallback_profile(signature)
+        profile["rarity"] = "common"
+        return profile
+
+    class CommonSpiritService:
+        async def mint_profile(self, signature, channel, anomalies, language="en"):
+            return _common_only(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():
+                yield "hi"
+
+            return gen()
+
+        def ambient_ready(self):
+            return False
+
+    monkeypatch.setattr(app.ws, "spirit_service", CommonSpiritService())
+    # If a drop were (incorrectly) rolled for a common-rarity summon, this
+    # would make it always hit — proving the rarity gate, not just the odds,
+    # is what's preventing a drop here.
+    monkeypatch.setattr(
+        app.ws, "roll_item_drop", lambda trigger: {"item_type": "x", "item_key": "y", "payload": {}}
+    )
+
+    token = _login(sync_client, "commoner")
+    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")
+        ws.send_json({"type": "summon"})
+        entity_frame = _read_until(ws, "entity")
+        assert entity_frame["entity"]["rarity"] == "common"
+        greeting = _read_until(ws, "utterance", kind="greeting")
+        assert greeting["kind"] == "greeting"
+
+    # Give the (already-awaited, so this should be immediate) reward path a
+    # moment to land, then assert no item was ever persisted for this user —
+    # the rarity gate, not just the roll odds, is what prevents a drop here.
+    for _ in range(20):
+        db_session.expire_all()
+        items = (
+            await db_session.execute(
+                select(InventoryItem).where(InventoryItem.user_id == user_id)
+            )
+        ).scalars().all()
+        await asyncio.sleep(0.02)
+    assert items == []