fix: reward guards now survive a reconnect

The three replay guards shipped earlier lived on the in-memory SeanceState,
which made them per-CONNECTION. I flagged that as an open residual at the
time: drop the socket and reconnect — or just open a second one — and the
client got a fresh empty guard and could be paid again for the same spirit.
summon_limiter bounded the rate of that, never the total.

`award_claims` is the durable form: one row per (seeker, presence,
milestone), with a UNIQUE constraint doing the actual enforcement. The claim
is a bare INSERT and losing the race raises IntegrityError, which is caught
and read as "already paid" — a check-then-insert would let two sockets both
read "unclaimed" and both pay. `crossing` is claimed by BOTH roads, so a
spirit crosses once whichever road arrives first.

Measured with the durable claim disabled: 5 reconnects paid 75 extra essence
on the ritual, 60 on a verdict, 140 on the passage, and two simultaneous
sockets paid 30 for one 15-essence ritual.

The four tests that were failing were the tests, not the guard. They compared
raw balances across reconnects, but re-opening a channel IS a summon, and
SUMMON_ESSENCE_TRICKLE is paid per summon by design (inventory.py:42, bounded
by summon_limiter rather than by any once-per-presence rule). The expected
trickle is now stated explicitly so the assertion speaks about the milestone
it is actually testing. Favor has no trickle, so it must not move at all —
asserted separately.

Anti-overshoot covered in both directions: a genuinely fresh presence still
pays in full across a reconnect, a corrected verdict still pays on a second
connection, and `test` stays freely repeatable since it never touches the
ledger.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/app/main.py

@@ -129,6 +129,45 @@ async def lifespan(app: FastAPI):
             "END IF; "
             "END $$;"
         ))
+
+        # The durable reward ledger (app/models/award_claim.py). `create_all`
+        # above already builds this table on a fresh database; these two
+        # statements are for an install that predates it — the CREATE is a
+        # no-op there, and the constraint is what actually enforces
+        # once-per-(seeker, presence, milestone), so it is asserted explicitly
+        # rather than left to whatever the table happened to be created with.
+        await conn.execute(text(
+            """
+            CREATE TABLE IF NOT EXISTS award_claims (
+                id UUID PRIMARY KEY,
+                user_id UUID NOT NULL REFERENCES users(id),
+                entity_id UUID NOT NULL REFERENCES entities(id),
+                award_key VARCHAR(64) NOT NULL,
+                claimed_at TIMESTAMPTZ NOT NULL DEFAULT now()
+            )
+            """
+        ))
+        await conn.execute(text(
+            "CREATE INDEX IF NOT EXISTS ix_award_claims_user_id ON award_claims (user_id)"
+        ))
+        await conn.execute(text(
+            "CREATE INDEX IF NOT EXISTS ix_award_claims_entity_id ON award_claims (entity_id)"
+        ))
+        # Same catalog-check shape as uq_unlocks_user_key above: `ADD
+        # CONSTRAINT` has no IF NOT EXISTS form. Unlike that one this
+        # constraint is not defense-in-depth — it IS the guard: _claim_award
+        # relies on the IntegrityError it raises to resolve two concurrent
+        # connections claiming the same award down to a single payout.
+        await conn.execute(text(
+            "DO $$ BEGIN "
+            "IF NOT EXISTS ("
+            "  SELECT 1 FROM pg_constraint WHERE conname = 'uq_award_claims_user_entity_key'"
+            ") THEN "
+            "  ALTER TABLE award_claims ADD CONSTRAINT uq_award_claims_user_entity_key "
+            "    UNIQUE (user_id, entity_id, award_key); "
+            "END IF; "
+            "END $$;"
+        ))
     cleanup_task = asyncio.create_task(_session_cleanup_loop())
     try:
         yield

backend/app/models/__init__.py

@@ -1,4 +1,5 @@
 from app.models.auth_session import AuthSession
+from app.models.award_claim import AwardClaim
 from app.models.contact_session import ContactSession
 from app.models.device import Device
 from app.models.entity import Entity
@@ -14,6 +15,7 @@ from app.models.waitlist_entry import WaitlistEntry
 __all__ = [
     "User",
     "AuthSession",
+    "AwardClaim",
     "ContactSession",
     "Device",
     "Entity",

backend/app/models/award_claim.py

@@ -0,0 +1,47 @@
+import uuid
+from datetime import datetime, timezone
+
+from sqlalchemy import DateTime, ForeignKey, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db import Base
+
+
+class AwardClaim(Base):
+    """One row per milestone a seeker has ACTUALLY been paid for a presence.
+
+    The three reward faucets in `app/ws.py` (the ritual milestone, a verdict,
+    a Passage beat) each used to be guarded by a set held on the in-memory
+    `SeanceState`. That guard is per-CONNECTION: a client that disconnects and
+    reconnects — or simply opens a second websocket — got a fresh, empty guard
+    and could be paid all over again for the same spirit. `summon_limiter`
+    bounded the rate of that, never the total.
+
+    This table is the durable form of those guards. The UNIQUE constraint on
+    (user_id, entity_id, award_key) is the actual enforcement, not a
+    belt-and-braces afterthought: `app.ws._claim_award` claims a milestone by
+    INSERTing here and treating an IntegrityError as "somebody already paid
+    this", which is what makes two connections racing the same award resolve
+    to exactly one payout instead of two.
+
+    `award_key` is the milestone within a presence, not a currency:
+    `ritual`, `judgment:<verdict>`, `passage:<layer>`, and `crossing` (claimed
+    by BOTH roads to a crossing — the Passage's `release` beat and the
+    `cross_over` verdict — so one spirit crosses once whichever road got there
+    first).
+    """
+
+    __tablename__ = "award_claims"
+    __table_args__ = (
+        UniqueConstraint(
+            "user_id", "entity_id", "award_key", name="uq_award_claims_user_entity_key"
+        ),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
+    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"), index=True)
+    entity_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("entities.id"), index=True)
+    award_key: Mapped[str] = mapped_column(String(64))
+    claimed_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )

backend/app/ws.py

@@ -55,6 +55,7 @@ from app.inventory import (
 )
 from app.llm.service import SpiritBusyError, spirit_service
 from app.models.auth_session import AuthSession, hash_token
+from app.models.award_claim import AwardClaim
 from app.models.contact_session import ContactSession
 from app.models.entity import Entity
 from app.models.entity_sighting import EntitySighting
@@ -153,31 +154,24 @@ class SeanceState:
     ritual_steps: int = 0
     ritual_completed: bool = False
     ritual_success: bool = False
-    # Whether the ritual milestone has ALREADY paid out for the current
+    # Awards this connection has already OBSERVED to be claimed, as
-    # presence. Exactly the same faucet `passage_paid` closes, on the other
+    # (entity_id, award_key) pairs — a pure read-through cache of rows in
-    # rite: `ritual_start` rewinds `ritual_completed` to False at any time
+    # `award_claims` (see `_claim_award`), never the guard itself.
-    # (and the UI offers precisely that button — RitualPanel's "attempt
+    #
-    # again", which honest play needs after a FAILED roll), so without this
+    # It exists only to spare the DB a doomed INSERT on the common replay
-    # a client could walk ritual_start + 4x ritual_step for +15 essence and
+    # path. It is safe precisely because it caches one direction: an entry
-    # an item roll, restart, and repeat forever off a single summon. The
+    # means "definitely already paid" (a fact that can never become false —
-    # limiter caps the cadence, not the total. So the milestone pays the
+    # rows are never deleted), and its ABSENCE means nothing at all, so a
-    # FIRST time it succeeds for a given presence and never again;
+    # miss always falls through to the database. It is deliberately never
-    # re-attempting still rolls, still reveals the traits on success, but
+    # cleared on a summon; a stale entry would only ever be correct.
-    # mints no new essence and rolls no new item. Cleared only on a fresh
+    #
-    # summon, never by `ritual_start` — that is the whole point.
+    # What used to live here instead — `ritual_paid: bool`,
-    ritual_paid: bool = False
+    # `judged_verdicts: set[str]`, `passage_paid: set[str]` — were the guards
-    # Which verdicts have ALREADY been applied to the ledger for the current
+    # themselves, and that was the hole: they are per-CONNECTION. Every one
-    # presence. Same class of hole: `judgment` had no once-per-presence
+    # of the faucets they close (see the comments on the handlers below)
-    # guard at all except for `cross_over`, so replaying {"type":
+    # reopened in full the moment a client dropped the socket and
-    # "judgment", "verdict": "trust"} against a benevolent spirit credited
+    # reconnected, or simply opened a second one alongside the first.
-    # CORRECT_JUDGMENT_ESSENCE *and* +0.05 favor *and* rolled an item on
+    awards_claimed: set[tuple[uuid.UUID, str]] = field(default_factory=set)
-    # every single frame — an unbounded faucet for both currencies (favor
-    # pins to +1.0, which then biases every future mint) that the 10/60s
-    # limiter only slowed down. Keyed by verdict rather than a single latch
-    # so an honest correction (a wrong `trust`, then the right `banish` on
-    # the same spirit) still resolves normally — what can never repeat is
-    # the SAME verdict on the SAME presence. Cleared only on a fresh summon.
-    judged_verdicts: set[str] = field(default_factory=set)
     # Workstream L (passage-doctrine spec): where the layered crossing rite
     # stands for the *current* entity. `passage_layer` is the next beat to
     # attempt, `passage_lied` remembers whether the layer just completed was
@@ -188,17 +182,6 @@ class SeanceState:
     passage_layer: str = passage.FIRST_LAYER
     passage_lied: bool = False
     passage_crossed: bool = False
-    # Which beats have ALREADY paid out for the current entity. Without
-    # this, essence is unbounded: `passage_start` rewinds the rite to
-    # `listen` at any time (and the UI offers exactly that button after a
-    # resisted release), so a client could walk listen/name/unbind/open for
-    # +28 essence, restart, and repeat forever off a single summon — the
-    # limiter caps the rate, not the total. A volatility collapse rewinds it
-    # the same way. So each layer pays the FIRST time it opens for a given
-    # presence and never again; re-walking it still reveals, still costs
-    # nothing already banked, but mints no new essence. Cleared only on a
-    # fresh summon, never by `passage_start` — that is the whole point.
-    passage_paid: set[str] = field(default_factory=set)
     # 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.
@@ -513,6 +496,46 @@ async def _summon(state: SeanceState, channel: str) -> tuple[Entity, bool]:
     raise last_error
 
 
+async def _claim_award(state: SeanceState, entity_id: str, award_key: str) -> bool:
+    """Claim a milestone for (this seeker, this presence). True exactly once.
+
+    This is the durable replacement for the three per-connection guards that
+    used to live on `SeanceState`. Every reward path in this file that is
+    meant to pay once per presence asks here first, and pays only on True.
+
+    The `award_claims` UNIQUE constraint is the real guard, not the SELECT
+    that never happens: the claim is a bare INSERT, and losing the race is an
+    IntegrityError, which we swallow and report as "already paid". That is
+    the only shape that holds when two websockets for the same account claim
+    the same award at the same instant — a check-then-insert would let both
+    read "unclaimed" and both pay. Same handling as `_summon`'s signature/name
+    collision above: catch, roll back, and take the winner's outcome as the
+    answer, rather than letting a 500 kill the séance.
+
+    Rows are never deleted, so a claim is permanent for that spirit — which is
+    the whole point. A genuinely NEW presence is a different `entity_id` and
+    therefore a fresh, unclaimed set of milestones; see the note in
+    `_summon_locked`.
+    """
+    key = (uuid.UUID(entity_id), award_key)
+    if key in state.awards_claimed:
+        return False  # fast path only; a miss still asks the database
+
+    async with session_maker() as db:
+        db.add(
+            AwardClaim(user_id=state.user_id, entity_id=key[0], award_key=award_key)
+        )
+        try:
+            await db.commit()
+        except IntegrityError:
+            await db.rollback()
+            state.awards_claimed.add(key)
+            return False
+
+    state.awards_claimed.add(key)
+    return True
+
+
 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
@@ -608,11 +631,18 @@ async def _summon_locked(state: SeanceState) -> None:
     state.ritual_steps = 0
     state.ritual_completed = False
     state.ritual_success = False
-    # Only a genuinely fresh summon re-opens the purse for the other two
+    # Nothing here re-opens the purse any more, and that is the fix.
-    # rites, exactly as `_reset_passage(new_entity=True)` does below.
+    #
-    state.ritual_paid = False
+    # A genuinely new presence still gets its own full purse — automatically,
-    state.judged_verdicts = set()
+    # because the purse is now keyed on `entity.id` in `award_claims` and a
-    _reset_passage(state, new_entity=True)
+    # new presence is a new row. What no longer re-opens it is a summon that
+    # lands back on the SAME spirit, and that distinction is the entire hole:
+    # `RETURN_CHANCE` means re-calling a known channel usually returns the
+    # familiar presence, so the old unconditional reset here let a client
+    # re-summon its way to an unlimited number of payouts for one spirit,
+    # bounded only by `summon_limiter`. Reconnecting did the same thing for
+    # free. Each spirit is worth its own rite — once.
+    _reset_passage(state)
     await state.send_queue.put(
         {"type": "entity", "entity": _public_entity(state.entity), "is_new": is_new}
     )
@@ -918,6 +948,11 @@ async def _handle_ritual_step(state: SeanceState, message: dict) -> None:
         return
 
     traits = state.entity.get("traits", {})
+    # Pinned before any await, for the same reason as in the judgment and
+    # Passage handlers below: the HTTP device-telemetry path drives the same
+    # SeanceState and can replace `state.entity` mid-handler, and the claim
+    # must land against the spirit whose rite this actually was.
+    entity_id = state.entity["id"]
     success = judgment.roll_ritual_success(traits)
     state.ritual_completed = True
     state.ritual_success = success
@@ -925,11 +960,12 @@ async def _handle_ritual_step(state: SeanceState, message: dict) -> None:
     await state.send_queue.put(
         {"type": "ritual_complete", "success": success, "revealed": revealed}
     )
-    # The milestone pays once per presence. A re-attempt after a rewind
+    # The milestone pays once per presence, for good — across rewinds
-    # (`ritual_start`) still rolls and still reveals on success — it just
+    # (`ritual_start`, the UI's "attempt again"), across reconnects, and
-    # doesn't mint a second payout. See `SeanceState.ritual_paid`.
+    # across two sockets racing each other. A re-attempt still rolls and
-    if success and not state.ritual_paid:
+    # still reveals on success; it just doesn't mint a second payout. See
-        state.ritual_paid = True
+    # `_claim_award`.
+    if success and await _claim_award(state, entity_id, "ritual"):
         await _reward_ritual_success(state)
 
 
@@ -971,14 +1007,28 @@ async def _handle_judgment(state: SeanceState, message: dict) -> None:
         ritual_success=state.ritual_success,
     )
 
-    # A verdict lands on a presence once. Replaying the same one — the UI
+    # A verdict lands on a presence once, durably. Replaying the same one —
-    # re-arms the panel after any result — re-reports the same reading for
+    # the UI re-arms the panel after any result, and a scripted client can
-    # free rather than paying it out again. See `judged_verdicts`; `test`
+    # simply reconnect or open a second socket — re-reports the same reading
-    # never touches the ledger so it is deliberately exempt and stays
+    # for free rather than paying it out again.
-    # freely repeatable.
+    #
-    already_judged = verdict != "test" and verdict in state.judged_verdicts
+    # Claimed PER VERDICT rather than as a single latch, so an honest
-    if verdict != "test":
+    # correction (a wrong `trust`, then the right `banish` on the same
-        state.judged_verdicts.add(verdict)
+    # spirit) still resolves normally; what can never repeat is the SAME
+    # verdict on the SAME presence. `test` never touches the ledger, so it is
+    # deliberately exempt and stays freely repeatable.
+    if verdict == "test":
+        already_judged = False
+    elif outcome.consequence == "crossed_over":
+        # One spirit, one crossing, whichever road got there first: the
+        # Passage's `release` beat claims this same key. Without the shared
+        # key, a seeker on two sockets could cross the one spirit down both
+        # roads and be paid for it twice.
+        already_judged = not await _claim_award(state, entity_id, "crossing")
+    else:
+        already_judged = not await _claim_award(
+            state, entity_id, f"judgment:{verdict}"
+        )
 
     # What will ACTUALLY reach the ledger. The frame below reports these,
     # not `outcome`'s face values: the client sums `essence_delta`/
@@ -1060,15 +1110,13 @@ async def _handle_judgment(state: SeanceState, message: dict) -> None:
 # uses), the `at_peace` write, and the frames.
 
 
-def _reset_passage(state: SeanceState, *, new_entity: bool = False) -> None:
+def _reset_passage(state: SeanceState) -> None:
+    """Rewind where the rite STANDS. It no longer touches what has been paid —
+    that lives in `award_claims` now and is keyed on the entity, so there is
+    nothing here a `passage_start` rewind could reopen."""
     state.passage_layer = passage.FIRST_LAYER
     state.passage_lied = False
     state.passage_crossed = False
-    if new_entity:
-        # Only a genuinely new presence re-opens the purse. A `passage_start`
-        # rewind must not, or the rite becomes an essence faucet (see
-        # `passage_paid` on SeanceState).
-        state.passage_paid = set()
 
 
 def _passage_frame(
@@ -1170,12 +1218,23 @@ async def _handle_passage_layer(state: SeanceState) -> None:
     if outcome.at_peace:
         state.passage_crossed = True
 
-    # A layer pays once per presence. Replaying it — via `passage_start`, via
+    # A layer pays once per presence, durably. Replaying it — via
-    # a collapse rewind, or via a hand-rolled client spamming the frame —
+    # `passage_start`, via a collapse rewind, via a reconnect, via a second
-    # reveals the same thing again for free rather than minting essence again.
+    # socket, or via a hand-rolled client spamming the frame — reveals the
-    award = outcome.essence if layer not in state.passage_paid else 0
+    # same thing again for free rather than minting essence again.
-    if outcome.result in ("opened", "crossed"):
+    award = 0
-        state.passage_paid.add(layer)
+    if outcome.result in ("opened", "crossed") and await _claim_award(
+        state, entity_id, f"passage:{layer}"
+    ):
+        award = outcome.essence
+
+    # The crossing itself (favor, `at_peace`, the item roll) is claimed
+    # separately and under a key the `cross_over` VERDICT claims too, so one
+    # spirit is paid for crossing exactly once no matter which road reached
+    # it — including two roads walked concurrently on two sockets.
+    crossing_paid = bool(
+        outcome.at_peace and await _claim_award(state, entity_id, "crossing")
+    )
 
     item = None
     if award or outcome.at_peace:
@@ -1188,7 +1247,7 @@ async def _handle_passage_layer(state: SeanceState) -> None:
             if user is not None:
                 if award:
                     credit_essence(user, award)
-                if outcome.at_peace:
+                if crossing_paid:
                     # Completing the Passage is the most compassionate act
                     # in the game, exactly as a correct cross_over verdict
                     # is — so it moves favor by the same amount.
@@ -1196,9 +1255,14 @@ async def _handle_passage_layer(state: SeanceState) -> None:
                         user.favor + judgment.FAVOR_CORRECT_CROSS_OVER
                     )
             if outcome.at_peace:
+                # Not gated on `crossing_paid`: the spirit really did cross,
+                # so the flag is set either way. It is only the PAYOUT that
+                # happens once. (When the claim was lost, the row is already
+                # at_peace and this write is a no-op.)
                 entity_row = await db.get(Entity, uuid.UUID(entity_id))
                 if entity_row is not None:
                     entity_row.at_peace = True
+            if crossing_paid:
                 item = roll_item_drop("judgment")
                 if item is not None:
                     db.add(

backend/tests/test_ws_passage.py

@@ -99,6 +99,32 @@ def _no_twists(monkeypatch):
     monkeypatch.setattr(app.ws, "veil_float", selective)
 
 
+def _force_new_presence(monkeypatch):
+    """Pin the summon draw so the NEXT summon mints a genuinely new spirit.
+
+    `_summon` draws against RETURN_CHANCE to decide whether the presence
+    already on this channel answers again, and re-calling the same channel
+    usually returns the SAME entity row — measured on this suite, 8 of 11
+    consecutive re-summons came back with `is_new: false` and an identical id.
+
+    That matters now that the purse is keyed on the entity (award_claims),
+    because "summon again" and "the same spirit answers again" are then two
+    different things. This test is about what a genuinely FRESH presence is
+    worth, so the draw is pinned rather than left to a coin flip; a draw of
+    1.0 is >= any possible `return_chance`, so the familiar presence never
+    answers. Layered over `_no_twists` (call it after), leaving every other
+    draw real.
+    """
+    real = app.ws.veil_float
+
+    def selective(entropy, context, *args, **kwargs):
+        if context == "answers":
+            return 1.0
+        return real(entropy, context, *args, **kwargs)
+
+    monkeypatch.setattr(app.ws, "veil_float", selective)
+
+
 # The exploit never reaches `release`. Crossing latches `passage_crossed`,
 # which correctly blocks replay — the faucet ran on the four beats BEFORE
 # release, rewound by the "walk it again" button after a resisted release.
@@ -260,17 +286,20 @@ async def test_a_genuinely_new_presence_reopens_the_purse(
 ):
     """The guard must not overshoot: each spirit is worth its own rite.
 
-    A `passage_paid` set that survived a summon would silently make every
+    A durable claim that survived a summon would silently make every spirit
-    spirit after the first worthless, which is a worse bug than the faucet.
+    after the first worthless, which is a worse bug than the faucet.
     """
     _no_twists(monkeypatch)
+    # The second summon must actually bring a DIFFERENT spirit — see
+    # `_force_new_presence`. Without this the test asserts a coin flip.
+    _force_new_presence(monkeypatch)
 
     token = _login(sync_client, "second-spirit")
     user_id = _user_id(sync_client, token)
 
     with _ws_connect(sync_client, token) as ws:
         _read_until(ws, "session")
-        _summon(ws)
+        first_entity = _summon(ws)
         baseline = await _essence(db_session, user_id)
 
         ws.send_json({"type": "passage_start"})
@@ -278,7 +307,11 @@ async def test_a_genuinely_new_presence_reopens_the_purse(
         _walk(ws, BEATS_BEFORE_RELEASE)
         after_first = await _essence(db_session, user_id)
 
-        _summon(ws)
+        second_entity = _summon(ws)
+        assert second_entity["entity"]["id"] != first_entity["entity"]["id"], (
+            "the second summon returned the same spirit — this test is about "
+            "a genuinely new presence"
+        )
         ws.send_json({"type": "passage_start"})
         _settle(ws)
         _walk(ws, BEATS_BEFORE_RELEASE)

backend/tests/test_ws_reward_durability.py

@@ -0,0 +1,682 @@
+"""WS integration tests for the DURABILITY of the reward guards in app.ws.
+
+`test_ws_passage.py` and `test_ws_reward_replay.py` closed three replay
+faucets — the ritual milestone, a verdict, and a Passage beat — but they
+closed them with sets held on the in-memory `SeanceState`, and every one of
+those tests replays the exploit down a SINGLE websocket. That is exactly the
+shape of the remaining hole: the guards were per-connection, so a client that
+dropped the socket and reconnected, or simply opened a second one alongside
+the first, got a fresh empty guard and could be paid all over again for the
+same spirit. Income stayed rate-bounded by `summon_limiter` (per user id,
+across connections) and unbounded in total.
+
+These tests reach past one connection:
+
+  * walk a rite to completion, DISCONNECT, reconnect on a new socket with the
+    same account, land back on the same spirit, re-walk it — the ledger must
+    not grow;
+  * two sockets open SIMULTANEOUSLY for the same account, both claiming the
+    same award — paid exactly once;
+  * the same race driven concurrently rather than interleaved, straight at
+    `_claim_award`, which is where the UNIQUE constraint on award_claims
+    actually decides the winner;
+  * and the anti-overshoot direction in both places: a genuinely fresh
+    presence still pays in full across a reconnect, and a corrected verdict
+    still pays.
+
+LANDING ON THE SAME SPIRIT ACROSS CONNECTIONS is the load-bearing detail. A
+spirit is matched by the *channel* signature, and with no anomalies at all
+that signature falls back to the CONTACT SESSION id — which is per-connection,
+so a bare reconnect+summon mints a brand new spirit and would prove nothing.
+So these tests feed the same anomaly stream a browser feeds (which is also
+what auto-summons), giving both connections the same channel, and pin the
+`answers` draw so the familiar presence reliably answers rather than 72% of
+the time.
+"""
+
+import asyncio
+import uuid
+
+import pytest
+from sqlalchemy import select
+
+import app.judgment as judgment_module
+import app.ws
+from app.entities import MIN_ANOMALIES_FOR_SIGNATURE, fallback_profile
+from app.inventory import (
+    CORRECT_JUDGMENT_ESSENCE,
+    RITUAL_SUCCESS_ESSENCE,
+    SUMMON_ESSENCE_TRICKLE,
+)
+from app.models.award_claim import AwardClaim
+from app.models.inventory_item import InventoryItem
+from app.models.user import User
+from app.rate_limit import RateLimiter
+
+
+class FakeSpiritService:
+    async def mint_profile(
+        self, signature, channel, anomalies, language="en", entropy=None, sky=None
+    ):
+        return fallback_profile(signature)
+
+    async def fragment(self, source, anomaly, language="en"):
+        return "listen"
+
+    async def wire_whisper(self, telemetry, language="en"):
+        return "the wire hums"
+
+    def chat_stream(self, entity, question, history, language="en"):
+        async def gen():
+            yield "here."
+
+        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)
+    # Module-level limiters are shared singletons accumulating real hit counts
+    # across the whole session (precedent: test_ws_reward_replay.py). These
+    # tests deliberately summon more than the real 4/60s cap allows, because
+    # the point is what happens on the SECOND connection — a rate limit there
+    # would mask the durability question instead of answering it.
+    for name in (
+        "summon_limiter",
+        "summon_ip_limiter",
+        "question_limiter",
+        "question_ip_limiter",
+        "fragment_limiter",
+        "fragment_ip_limiter",
+        "ritual_limiter",
+        "ritual_ip_limiter",
+        "judgment_limiter",
+        "judgment_ip_limiter",
+        "passage_limiter",
+        "passage_ip_limiter",
+    ):
+        monkeypatch.setattr(app.ws, name, RateLimiter(max_requests=10_000, window_seconds=60))
+
+
+def _read_until(ws, msg_type, max_frames=120, **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 _user_id(sync_client, token):
+    return uuid.UUID(
+        sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
+    )
+
+
+def _ws_connect(sync_client, token):
+    return sync_client.websocket_connect(
+        "/ws/session", headers={"cookie": f"qm_session={token}"}
+    )
+
+
+def _settle(ws):
+    """Force a full round trip so any post-frame DB write has landed. The
+    connection's message loop is sequential, so a pong is a hard guarantee
+    that the previous handler returned — not a poll-and-hope."""
+    ws.send_json({"type": "ping"})
+    _read_until(ws, "pong")
+
+
+# One fixed anomaly pattern = one fixed channel signature, shared by every
+# connection in a test. `signature_from_anomalies` buckets frequency by digit
+# count and sorts magnitudes, so identical frames give an identical channel
+# regardless of which contact session sent them.
+_CHANNEL = [
+    {"type": "anomaly", "source": "emf", "frequency": 121.5, "magnitude": 30.0},
+    {"type": "anomaly", "source": "emf", "frequency": 122.5, "magnitude": 31.0},
+    {"type": "anomaly", "source": "emf", "frequency": 123.5, "magnitude": 32.0},
+]
+assert len(_CHANNEL) >= MIN_ANOMALIES_FOR_SIGNATURE
+
+
+def _familiar_answers(monkeypatch):
+    """Pin the summon draw so a known channel ALWAYS returns its familiar
+    presence.
+
+    Untouched, this is a genuine draw against RETURN_CHANCE — roughly 7 times
+    in 10. These tests need the two connections to be looking at the SAME
+    spirit for the question ("can it be paid twice?") to mean anything, so a
+    draw of 0.0 is used, which is below any possible `return_chance`. Every
+    other draw stays real.
+    """
+    real = app.ws.veil_float
+
+    def selective(entropy, context, *args, **kwargs):
+        if context == "answers":
+            return 0.0
+        return real(entropy, context, *args, **kwargs)
+
+    monkeypatch.setattr(app.ws, "veil_float", selective)
+
+
+def _open_channel(ws):
+    """Drive the channel a browser drives: enough anomalies to fingerprint it,
+    which is itself what triggers the summon (see `_handle_anomaly`)."""
+    _read_until(ws, "session")
+    for frame in _CHANNEL:
+        ws.send_json(frame)
+    entity = _read_until(ws, "entity")
+    _settle(ws)
+    return entity["entity"]["id"]
+
+
+def _run_ritual(ws, steps=4):
+    """One full attempt: the rewind button, then every step."""
+    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")
+    _settle(ws)
+    return result
+
+
+def _judge(ws, verdict):
+    ws.send_json({"type": "judgment", "verdict": verdict})
+    result = _read_until(ws, "judgment_result")
+    _settle(ws)
+    return result
+
+
+async def _ledger(db_session, user_id):
+    db_session.expire_all()
+    user = await db_session.get(User, user_id)
+    items = (
+        await db_session.execute(
+            select(InventoryItem).where(InventoryItem.user_id == user_id)
+        )
+    ).scalars().all()
+    return user.essence, user.favor, len(items)
+
+
+def _always_wins(monkeypatch):
+    monkeypatch.setattr(
+        app.ws.judgment, "roll_ritual_success", lambda traits, rng=None: True
+    )
+
+
+def _verdict_outcomes(monkeypatch, table):
+    monkeypatch.setattr(
+        app.ws.judgment, "judge_verdict", lambda verdict, traits, **kw: table[verdict]
+    )
+
+
+_CORRECT_TRUST = judgment_module.JudgmentOutcome(
+    True, judgment_module.FAVOR_CORRECT_TRUST, CORRECT_JUDGMENT_ESSENCE, False, "reward"
+)
+_WRONG_TRUST = judgment_module.JudgmentOutcome(
+    False, judgment_module.FAVOR_WRONG_TRUST, 0, False, "escalation"
+)
+_CORRECT_BANISH = judgment_module.JudgmentOutcome(
+    True, judgment_module.FAVOR_CORRECT_BANISH, CORRECT_JUDGMENT_ESSENCE, False, "reward"
+)
+
+
+# --- reconnecting ----------------------------------------------------------
+
+
+# How many times each reconnect test re-opens the socket. Named because the
+# expected summon trickle is derived from it, and a bare 5 in two places that
+# must agree is exactly how that arithmetic drifts.
+RECONNECTS = 5
+
+
+@pytest.mark.asyncio
+async def test_reconnecting_does_not_reopen_the_ritual_purse(
+    sync_client, db_session, monkeypatch
+):
+    """The exploit the in-memory guard could not see, executed literally.
+
+    Complete the rite, DROP THE SOCKET, reconnect on a fresh websocket with
+    the same account, tune back to the same channel so the same spirit
+    answers, and complete the rite again — five times over. The seeker must
+    finish with exactly one milestone's worth of essence, because there was
+    only ever one spirit.
+
+    `ritual_paid` was a bool on `SeanceState`, which dies with the connection,
+    so before this every reconnect paid RITUAL_SUCCESS_ESSENCE again and
+    rolled another item.
+    """
+    _always_wins(monkeypatch)
+    _familiar_answers(monkeypatch)
+
+    token = _login(sync_client, "ritual-reconnector")
+    user_id = _user_id(sync_client, token)
+
+    with _ws_connect(sync_client, token) as ws:
+        first_entity = _open_channel(ws)
+        baseline, _, _ = await _ledger(db_session, user_id)
+        assert _run_ritual(ws)["success"] is True
+    after_first, _, items_after_first = await _ledger(db_session, user_id)
+
+    for _ in range(RECONNECTS):
+        with _ws_connect(sync_client, token) as ws:
+            entity_id = _open_channel(ws)
+            assert entity_id == first_entity, (
+                "the reconnect landed on a different spirit — this test cannot "
+                "say anything about paying the same one twice"
+            )
+            assert _run_ritual(ws)["success"] is True, "the replay did not even complete"
+
+    after_reconnects, _, items_after_reconnects = await _ledger(db_session, user_id)
+
+    assert after_first - baseline == RITUAL_SUCCESS_ESSENCE, (
+        "the rite paid nothing — the test proves nothing"
+    )
+    # Each reconnect re-opens the channel, and opening a channel is a summon,
+    # which pays SUMMON_ESSENCE_TRICKLE every time by design (app/inventory.py:42
+    # — "every successful summon, any mode"; it is bounded by summon_limiter at
+    # 4/60s, not by any once-per-presence rule). That trickle is NOT what this
+    # test is about, so it is expected explicitly rather than folded into the
+    # comparison: what must not repeat is the 15-essence ritual MILESTONE.
+    expected_trickle = RECONNECTS * SUMMON_ESSENCE_TRICKLE
+    assert after_reconnects == after_first + expected_trickle, (
+        f"reconnecting minted {after_reconnects - after_first - expected_trickle} extra "
+        f"essence beyond the {expected_trickle} of expected summon trickle, across "
+        f"{RECONNECTS} fresh websockets; the guard dies with the connection"
+    )
+    assert items_after_reconnects == items_after_first, (
+        f"reconnecting rolled {items_after_reconnects - items_after_first} extra item "
+        "drops off a single spirit"
+    )
+
+
+@pytest.mark.asyncio
+async def test_reconnecting_does_not_reopen_the_judgment_purse(
+    sync_client, db_session, monkeypatch
+):
+    """Same hole, the verdict road. `judged_verdicts` was a set on
+    `SeanceState`; every reconnect handed the client an empty one, so the same
+    verdict on the same spirit paid essence AND favor AND an item roll again,
+    with favor walking toward its +1.0 ceiling (which then biases every future
+    mint)."""
+    _verdict_outcomes(monkeypatch, {"trust": _CORRECT_TRUST})
+    _familiar_answers(monkeypatch)
+
+    token = _login(sync_client, "judgment-reconnector")
+    user_id = _user_id(sync_client, token)
+
+    with _ws_connect(sync_client, token) as ws:
+        first_entity = _open_channel(ws)
+        baseline_essence, baseline_favor, _ = await _ledger(db_session, user_id)
+        _judge(ws, "trust")
+    essence_one, favor_one, items_one = await _ledger(db_session, user_id)
+
+    replays = []
+    for _ in range(RECONNECTS):
+        with _ws_connect(sync_client, token) as ws:
+            assert _open_channel(ws) == first_entity
+            replays.append(_judge(ws, "trust"))
+
+    essence_end, favor_end, items_end = await _ledger(db_session, user_id)
+
+    assert essence_one - baseline_essence == CORRECT_JUDGMENT_ESSENCE
+    assert favor_one - baseline_favor == pytest.approx(judgment_module.FAVOR_CORRECT_TRUST)
+    # As in the ritual test: re-opening the channel is a summon, and the
+    # trickle is paid per summon by design. The VERDICT is what must not repeat.
+    # Favor, checked just below, has no trickle — so it must not move at all.
+    expected_trickle = RECONNECTS * SUMMON_ESSENCE_TRICKLE
+    assert essence_end == essence_one + expected_trickle, (
+        f"reconnecting minted {essence_end - essence_one - expected_trickle} extra "
+        f"essence beyond the {expected_trickle} of expected summon trickle, across "
+        f"{RECONNECTS} fresh websockets; the verdict guard dies with the connection"
+    )
+    assert favor_end == pytest.approx(favor_one), (
+        f"reconnecting moved favor by a further {favor_end - favor_one}; a client "
+        "that reconnects in a loop can pin favor at the +1.0 ceiling"
+    )
+    assert items_end == items_one
+    for frame in replays:
+        assert frame["essence_delta"] == 0 and frame["favor_delta"] == 0, (
+            "a verdict replayed on a fresh connection advertised essence/favor "
+            "it did not pay"
+        )
+
+
+@pytest.mark.asyncio
+async def test_reconnecting_does_not_reopen_the_passage_purse(
+    sync_client, db_session, monkeypatch
+):
+    """Same hole, the Passage road. `passage_paid` was a set on `SeanceState`,
+    so every reconnect re-credited the whole ladder of beats."""
+    real = app.ws.veil_float
+
+    def selective(entropy, context, *args, **kwargs):
+        # No collapses, no lies — the totals must depend on the guard, not the
+        # draw — and the familiar presence always answers.
+        if context.startswith("passage:"):
+            return 1.0
+        if context == "answers":
+            return 0.0
+        return real(entropy, context, *args, **kwargs)
+
+    monkeypatch.setattr(app.ws, "veil_float", selective)
+
+    # Every beat but `release`: crossing latches `at_peace`, which correctly
+    # takes the spirit out of reach of any later summon.
+    beats = len(app.ws.passage.LAYERS) - 1
+
+    token = _login(sync_client, "passage-reconnector")
+    user_id = _user_id(sync_client, token)
+
+    def _walk(ws):
+        for _ in range(beats):
+            ws.send_json({"type": "passage_layer"})
+            _read_until(ws, "passage_result")
+        _settle(ws)
+
+    with _ws_connect(sync_client, token) as ws:
+        first_entity = _open_channel(ws)
+        baseline, _, _ = await _ledger(db_session, user_id)
+        ws.send_json({"type": "passage_start"})
+        _settle(ws)
+        _walk(ws)
+    after_first, _, _ = await _ledger(db_session, user_id)
+
+    for _ in range(RECONNECTS):
+        with _ws_connect(sync_client, token) as ws:
+            assert _open_channel(ws) == first_entity
+            ws.send_json({"type": "passage_start"})
+            _settle(ws)
+            _walk(ws)
+
+    after_reconnects, _, _ = await _ledger(db_session, user_id)
+
+    assert after_first - baseline > 0, "the rite paid nothing — the test proves nothing"
+    # Re-opening the channel summons, and the trickle is per-summon by design;
+    # the BEATS are what must not pay twice.
+    expected_trickle = RECONNECTS * SUMMON_ESSENCE_TRICKLE
+    assert after_reconnects == after_first + expected_trickle, (
+        f"reconnecting minted {after_reconnects - after_first - expected_trickle} extra "
+        f"essence beyond the {expected_trickle} of expected summon trickle, across "
+        f"{RECONNECTS} fresh websockets; the passage is an unbounded faucet again"
+    )
+
+
+# --- two sockets at once ---------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_two_simultaneous_connections_pay_the_ritual_once(
+    sync_client, db_session, monkeypatch
+):
+    """Not a reconnect: both sockets are OPEN AT THE SAME TIME.
+
+    This is the cheaper half of the exploit — no disconnect needed, just a
+    second tab. Each connection has its own `SeanceState` and therefore had
+    its own empty guard, so the two of them paid the same spirit's ritual
+    twice while every replay test in the suite watched a single socket.
+    """
+    _always_wins(monkeypatch)
+    _familiar_answers(monkeypatch)
+
+    token = _login(sync_client, "ritual-two-tabs")
+    user_id = _user_id(sync_client, token)
+
+    with _ws_connect(sync_client, token) as first, _ws_connect(sync_client, token) as second:
+        entity_a = _open_channel(first)
+        entity_b = _open_channel(second)
+        assert entity_a == entity_b, (
+            "the two connections landed on different spirits — this test cannot "
+            "say anything about paying the same one twice"
+        )
+        baseline, _, items_baseline = await _ledger(db_session, user_id)
+
+        # Interleaved, so neither connection's rite is finished before the
+        # other's begins.
+        first.send_json({"type": "ritual_start"})
+        second.send_json({"type": "ritual_start"})
+        for i in range(1, 5):
+            first.send_json({"type": "ritual_step", "step": i})
+            second.send_json({"type": "ritual_step", "step": i})
+        assert _read_until(first, "ritual_complete")["success"] is True
+        assert _read_until(second, "ritual_complete")["success"] is True
+        _settle(first)
+        _settle(second)
+
+        essence_end, _, items_end = await _ledger(db_session, user_id)
+
+    assert essence_end - baseline == RITUAL_SUCCESS_ESSENCE, (
+        f"two simultaneous connections paid {essence_end - baseline} for one "
+        f"spirit's ritual instead of {RITUAL_SUCCESS_ESSENCE}"
+    )
+    assert items_end - items_baseline <= 1, (
+        "two simultaneous connections each rolled the ritual's item drop"
+    )
+
+
+@pytest.mark.asyncio
+async def test_two_simultaneous_connections_pay_a_verdict_once(
+    sync_client, db_session, monkeypatch
+):
+    _verdict_outcomes(monkeypatch, {"trust": _CORRECT_TRUST})
+    _familiar_answers(monkeypatch)
+
+    token = _login(sync_client, "judgment-two-tabs")
+    user_id = _user_id(sync_client, token)
+
+    with _ws_connect(sync_client, token) as first, _ws_connect(sync_client, token) as second:
+        assert _open_channel(first) == _open_channel(second)
+        baseline_essence, baseline_favor, _ = await _ledger(db_session, user_id)
+
+        first.send_json({"type": "judgment", "verdict": "trust"})
+        second.send_json({"type": "judgment", "verdict": "trust"})
+        frames = [
+            _read_until(first, "judgment_result"),
+            _read_until(second, "judgment_result"),
+        ]
+        _settle(first)
+        _settle(second)
+
+        essence_end, favor_end, _ = await _ledger(db_session, user_id)
+
+    assert essence_end - baseline_essence == CORRECT_JUDGMENT_ESSENCE, (
+        f"two simultaneous connections paid {essence_end - baseline_essence} for one "
+        f"verdict instead of {CORRECT_JUDGMENT_ESSENCE}"
+    )
+    assert favor_end - baseline_favor == pytest.approx(
+        judgment_module.FAVOR_CORRECT_TRUST
+    ), "two simultaneous connections moved favor twice for one verdict"
+    # And the frames must be honest about it: exactly one of them paid.
+    assert sum(f["essence_delta"] for f in frames) == essence_end - baseline_essence
+
+
+@pytest.mark.asyncio
+async def test_a_concurrent_claim_race_resolves_to_one_winner(sync_client, db_session):
+    """Straight at the constraint, with real concurrency.
+
+    The two-tabs tests above interleave frames but the server still handles
+    them one at a time, so they prove the guard is DURABLE without ever
+    forcing the two claims to be in flight together. This one does: twelve
+    coroutines claim the same (user, entity, award) at once via
+    `asyncio.gather`.
+
+    Exactly one may win. If `_claim_award` were a check-then-insert, several
+    would read "unclaimed" and all of them would pay; if it did not catch the
+    IntegrityError, the losers would raise instead of returning False and
+    would take down their whole séance with a 500.
+    """
+    token = _login(sync_client, "claim-racer")
+    user_id = _user_id(sync_client, token)
+
+    with _ws_connect(sync_client, token) as ws:
+        entity_id = _open_channel(ws)
+
+    state = app.ws.SeanceState(
+        user_id=user_id, session_id=uuid.uuid4(), client_ip="127.0.0.1"
+    )
+
+    async def claim():
+        # A private state per racer: sharing one would let the in-memory fast
+        # path answer some of them, which is exactly what must NOT decide this.
+        racer = app.ws.SeanceState(
+            user_id=user_id, session_id=state.session_id, client_ip="127.0.0.1"
+        )
+        return await app.ws._claim_award(racer, entity_id, "ritual")
+
+    results = await asyncio.gather(*(claim() for _ in range(12)))
+
+    assert sum(results) == 1, (
+        f"{sum(results)} of 12 concurrent claims won the same award; the "
+        "UNIQUE constraint is not deciding the race"
+    )
+    db_session.expire_all()
+    rows = (
+        await db_session.execute(
+            select(AwardClaim).where(
+                AwardClaim.user_id == user_id,
+                AwardClaim.entity_id == uuid.UUID(entity_id),
+                AwardClaim.award_key == "ritual",
+            )
+        )
+    ).scalars().all()
+    assert len(rows) == 1, f"the race left {len(rows)} claim rows for one award"
+
+
+# --- the anti-overshoot direction ------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_a_fresh_presence_after_a_reconnect_still_pays_in_full(
+    sync_client, db_session, monkeypatch
+):
+    """The guard must not overshoot across connections either.
+
+    A durable claim that keyed on the user alone — or on anything coarser than
+    the presence — would silently make every spirit after the first worthless
+    for the rest of that account's life, which is a far worse bug than the
+    faucet. Reconnect onto a DIFFERENT channel, meet a different spirit, and
+    the rite must pay in full again.
+    """
+    _always_wins(monkeypatch)
+
+    token = _login(sync_client, "fresh-after-reconnect")
+    user_id = _user_id(sync_client, token)
+
+    with _ws_connect(sync_client, token) as ws:
+        first_entity = _open_channel(ws)
+        baseline, _, _ = await _ledger(db_session, user_id)
+        _run_ritual(ws)
+    after_first, _, _ = await _ledger(db_session, user_id)
+
+    # A different anomaly pattern is a different channel, so a different
+    # presence answers — the honest version of what the replay tests fake.
+    with _ws_connect(sync_client, token) as ws:
+        _read_until(ws, "session")
+        for i in range(MIN_ANOMALIES_FOR_SIGNATURE):
+            ws.send_json(
+                {"type": "anomaly", "source": "radio", "frequency": 9000.0 + i, "magnitude": 77.0 + i}
+            )
+        second_entity = _read_until(ws, "entity")["entity"]["id"]
+        _settle(ws)
+        assert second_entity != first_entity, (
+            "the second channel produced the same spirit — this test is about a "
+            "genuinely new presence"
+        )
+        _run_ritual(ws)
+    after_second, _, _ = await _ledger(db_session, user_id)
+
+    first_rite = after_first - baseline
+    assert first_rite == RITUAL_SUCCESS_ESSENCE
+    # The second connection's summon pays its own trickle too, so compare only
+    # the amount that arrived once the rite completed.
+    assert after_second - after_first >= RITUAL_SUCCESS_ESSENCE, (
+        "a fresh presence met on a new connection did not re-open the purse — "
+        "every spirit after the first is worth nothing"
+    )
+
+
+@pytest.mark.asyncio
+async def test_a_correction_still_pays_on_a_second_connection(
+    sync_client, db_session, monkeypatch
+):
+    """The claim is keyed PER VERDICT, and that must survive the move to the
+    database. A seeker who calls it wrong on one connection and right on the
+    next is correcting themselves, not replaying — a blanket "judged once per
+    spirit" row would make the corrected verdict a no-op."""
+    _verdict_outcomes(monkeypatch, {"trust": _WRONG_TRUST, "banish": _CORRECT_BANISH})
+    _familiar_answers(monkeypatch)
+
+    token = _login(sync_client, "corrector-reconnect")
+    user_id = _user_id(sync_client, token)
+
+    with _ws_connect(sync_client, token) as ws:
+        first_entity = _open_channel(ws)
+        baseline_essence, baseline_favor, _ = await _ledger(db_session, user_id)
+        wrong = _judge(ws, "trust")
+
+    with _ws_connect(sync_client, token) as ws:
+        assert _open_channel(ws) == first_entity
+        right = _judge(ws, "banish")
+
+    essence_end, favor_end, _ = await _ledger(db_session, user_id)
+
+    assert wrong["consequence"] == "escalation"
+    assert right["consequence"] == "reward"
+    # One reconnect here, so exactly one summon trickle rides along with the
+    # corrected verdict's payout.
+    assert essence_end - baseline_essence == CORRECT_JUDGMENT_ESSENCE + SUMMON_ESSENCE_TRICKLE, (
+        "the corrected verdict paid nothing on the second connection — the "
+        "durable guard is too broad"
+    )
+    assert favor_end - baseline_favor == pytest.approx(
+        judgment_module.FAVOR_WRONG_TRUST + judgment_module.FAVOR_CORRECT_BANISH
+    )
+
+
+@pytest.mark.asyncio
+async def test_the_test_verdict_stays_freely_repeatable_across_connections(
+    sync_client, db_session, monkeypatch
+):
+    """`test` is a diagnostic pulse that never touches the ledger, so it is
+    deliberately exempt — it must never write a claim row, or the panel would
+    stop answering a seeker who pulses the same spirit from a second tab."""
+    _familiar_answers(monkeypatch)
+
+    token = _login(sync_client, "tester-reconnect")
+    user_id = _user_id(sync_client, token)
+
+    results = []
+    for _ in range(3):
+        with _ws_connect(sync_client, token) as ws:
+            entity_id = _open_channel(ws)
+            baseline, baseline_favor, _ = await _ledger(db_session, user_id)
+            results.append(_judge(ws, "test"))
+            essence_end, favor_end, _ = await _ledger(db_session, user_id)
+            assert essence_end == baseline and favor_end == baseline_favor
+
+    assert all(r["consequence"] == "neutral" for r in results)
+    db_session.expire_all()
+    claims = (
+        await db_session.execute(
+            select(AwardClaim).where(
+                AwardClaim.user_id == user_id,
+                AwardClaim.entity_id == uuid.UUID(entity_id),
+            )
+        )
+    ).scalars().all()
+    assert claims == [], "the `test` verdict wrote a claim row and is no longer free"

backend/tests/test_ws_reward_replay.py

@@ -192,6 +192,31 @@ def _verdict_outcomes(monkeypatch, table):
     )
 
 
+def _force_new_presence(monkeypatch):
+    """Pin the summon draw so the NEXT summon mints a genuinely new spirit.
+
+    `_summon` draws against RETURN_CHANCE to decide whether the presence
+    already on this channel answers again, and re-calling the same channel
+    usually returns the SAME entity row — measured on this suite, 8 of 11
+    consecutive re-summons came back with `is_new: false` and an identical id.
+
+    That matters now that the purse is keyed on the entity (award_claims),
+    because "summon again" and "the same spirit answers again" are then two
+    different things. These tests are about what a genuinely FRESH presence is
+    worth, so the draw is pinned rather than left to a coin flip; a draw of
+    1.0 is >= any possible `return_chance`, so the familiar presence never
+    answers. Every other draw stays real.
+    """
+    real = app.ws.veil_float
+
+    def selective(entropy, context, *args, **kwargs):
+        if context == "answers":
+            return 1.0
+        return real(entropy, context, *args, **kwargs)
+
+    monkeypatch.setattr(app.ws, "veil_float", selective)
+
+
 # --- the ritual faucet -----------------------------------------------------
 
 
@@ -272,22 +297,29 @@ async def test_a_genuinely_new_presence_reopens_the_ritual_purse(
 ):
     """The guard must not overshoot: each spirit is worth its own rite.
 
-    A `ritual_paid` latch that survived a summon would silently make every
+    A durable claim that survived a summon would silently make every spirit
-    spirit after the first worthless.
+    after the first worthless.
     """
     _always_wins(monkeypatch)
+    # The second summon must actually bring a DIFFERENT spirit — see
+    # `_force_new_presence`. Without this the test asserts a coin flip.
+    _force_new_presence(monkeypatch)
 
     token = _login(sync_client, "ritual-second-spirit")
     user_id = _user_id(sync_client, token)
 
     with _ws_connect(sync_client, token) as ws:
         _read_until(ws, "session")
-        _summon(ws)
+        first_entity = _summon(ws)
         baseline, _, _ = await _ledger(db_session, user_id)
         _run_ritual(ws)
         after_first, _, _ = await _ledger(db_session, user_id)
 
-        _summon(ws)
+        second_entity = _summon(ws)
+        assert second_entity["entity"]["id"] != first_entity["entity"]["id"], (
+            "the second summon returned the same spirit — this test is about "
+            "a genuinely new presence"
+        )
         _run_ritual(ws)
         after_second, _, _ = await _ledger(db_session, user_id)
 
@@ -425,17 +457,24 @@ async def test_a_genuinely_new_presence_reopens_the_judgment_purse(
     sync_client, db_session, monkeypatch
 ):
     _verdict_outcomes(monkeypatch, {"trust": _CORRECT_TRUST})
+    # The second summon must actually bring a DIFFERENT spirit — see
+    # `_force_new_presence`. Without this the test asserts a coin flip.
+    _force_new_presence(monkeypatch)
 
     token = _login(sync_client, "judgment-second-spirit")
     user_id = _user_id(sync_client, token)
 
     with _ws_connect(sync_client, token) as ws:
         _read_until(ws, "session")
-        _summon(ws)
+        first_entity = _summon(ws)
         _judge(ws, "trust")
         after_first, _, _ = await _ledger(db_session, user_id)
 
-        _summon(ws)
+        second_entity = _summon(ws)
+        assert second_entity["entity"]["id"] != first_entity["entity"]["id"], (
+            "the second summon returned the same spirit — this test is about "
+            "a genuinely new presence"
+        )
         _judge(ws, "trust")
         after_second, _, _ = await _ledger(db_session, user_id)