"""Phase 2 (the information war): true/false tips, heat, Magistrate Heron. GPU-free and deterministic. Mechanics tests drive the engine with a no-op policy (so trading never perturbs prices) and control prices / positions directly. The truth-firewall test reuses the StubClient pattern (one valid JSON decision per convo) to inspect each creature's prompt and prove the tip's hidden truth never reaches a single message. Run: python -m pytest tests/ -q """ import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from ttw import moves from ttw.agents import make_council_policy from ttw.council import CREATURE_ENGINE from ttw.dummy import make_random_policy from ttw.game import Game from ttw.world import ( EXILE_CHAPTER_TURNS, INVESTIGATION_THRESHOLD, ShortPosition, ) def _noop_policy(world, name): return [], [] def _game(policy=None, seed=1): return Game(policy or _noop_policy, deck_seed=seed) # --- StubClient (mirrors tests/test_council_policy.py) ------------------------ class StubClient: """Fake engine handle: one valid JSON decision per convo, echoing the creature name (parsed from the system prompt) into its thought.""" def __init__(self, engine_id: str): self.engine_id = engine_id def chat_batch(self, conversations, max_tokens=256, temperature=0.7): out = [] for convo in conversations: system = convo[0]["content"] name = system.split("You are ", 1)[1].split(",", 1)[0] out.append( json.dumps( { "thought": f"{name} via {self.engine_id}", "offers": [{"side": "sell", "good": "honey", "price": 9, "qty": 1}], "gossip": "", } ) ) return out def _clients(): return {eid: StubClient(eid) for eid in set(CREATURE_ENGINE.values())} # --- (1) true tip + front-running short pays, strictly beats the false control - def test_true_tip_short_settles_positive_and_beats_false_control(): # TRUE: the tip names the legend's good (the truth), so the short sits on the # good the legend will crash and settles profitably. g = _game() legend = g.deck.peek() good = legend.affected_good g.world.patron.shorts.append(ShortPosition(good=good, qty=4, entry_price=20.0)) g.queue_player_move("tip", target="Fenn", text="it will crash", truth=True) g.tempt_fate() g.world.last_price[good] = 5.0 # the legend's crash, held by the no-op policy g.step() true_pnl = g.world.patron.realized_pnl assert true_pnl > 0 assert all(not s.open for s in g.world.patron.shorts) # the bound short settled # FALSE control: the lie names the wrong good, so the short the player places # on it is NOT on the legend's good and never settles -> no edge. other = next(x for x in g.world.last_price if x != good) h = _game() h.world.patron.shorts.append(ShortPosition(good=other, qty=4, entry_price=20.0)) h.queue_player_move("tip", target="Fenn", good=other, text="it will crash", truth=False) h.tempt_fate() h.world.last_price[other] = 5.0 h.step() false_pnl = h.world.patron.realized_pnl assert false_pnl == 0 assert true_pnl > false_pnl # --- (2) TRUTH FIREWALL: truth/legend_key never reach any prompt -------------- def test_truth_firewall_truth_and_legend_key_never_in_any_prompt(): g = Game(make_council_policy(_clients(), temperature=0.7), deck_seed=1) g.queue_player_move("tip", target="Fenn", text="the honey crop will fail", truth=True) g.step() messages = g.policy.state["messages"] assert messages # the batch ran banned = ("true", "false", "truth", "legend_key") for name, convo in messages.items(): blob = json.dumps(convo).lower() for token in banned: assert token not in blob, f"{token!r} leaked into {name}'s prompt" # The tip TEXT itself does reach the targeted creature (the legible signal). fenn_prompt = json.dumps(messages["Fenn"]) assert "the honey crop will fail" in fenn_prompt # --- (3) heat math: true tip/legend tells raise heat; false tip adds none ------ def _crash_policy(good): """A policy whose only intent is an unmatched sell of `good`: with no buyer it is pure residual supply, so drift_prices pushes that good's price DOWN this turn (a deterministic in-turn crash for the suspicious-win check).""" from ttw.actions import Offer def policy(world, name): if name == "Oona": # Oona is the honey-maker; one big sell, no buyers return [Offer(name, "sell", good, max(1, int(world.last_price[good])), 30)], [] return [], [] return policy def test_heat_rises_on_suspicious_wins_and_false_tip_adds_none(): good = "honey" # Case (a): true recent tip + a price drop on an open short -> +15 heat. g = _game(_crash_policy(good)) g.world.patron.tips.append( moves.Tip(target="Fenn", good=good, legend_key="", truth=True, turn_planted=g.world.turn) ) g.world.patron.shorts.append(ShortPosition(good=good, qty=2, entry_price=9.0)) heat0 = g.world.patron.heat g.step() assert g.world.last_price[good] < 9.0 # the crash landed this turn assert g.world.patron.heat == heat0 + moves.HEAT_SHORT_INTO_TIPPED_CRASH # FALSE tip: same drop, same short, but the tip is a lie -> no heat from it. h = _game(_crash_policy(good)) h.world.patron.tips.append( moves.Tip(target="Fenn", good=good, legend_key="", truth=False, turn_planted=h.world.turn) ) h.world.patron.shorts.append(ShortPosition(good=good, qty=2, entry_price=9.0)) heat0h = h.world.patron.heat h.step() assert h.world.last_price[good] < 9.0 # same crash assert h.world.patron.heat == heat0h # but a false tip raised no heat def test_short_before_legend_adds_twenty_heat(): g = _game() legend = g.deck.peek() good = legend.affected_good g.world.patron.shorts.append(ShortPosition(good=good, qty=3, entry_price=20.0)) g.tempt_fate() heat0 = g.world.patron.heat g.world.last_price[good] = 5.0 g.step() assert g.world.patron.heat == heat0 + moves.HEAT_SHORT_BEFORE_LEGEND def test_run_on_hoard_crashes_good_at_settlement_policy_independent(): """A panic legend (crash_mult < 1) crashes its good at settlement, AFTER the market clears, so the crash holds no matter how the creatures traded. This is the load-bearing fix: the insider short pays even against a live LLM market that never chose to sell. Uses a one-legend deck so the draw is deterministic. """ from ttw.events import EVENTS, EventDeck run = next(e for e in EVENTS if e.key == "run_on_hoard") # crash_mult=0.5, honey # (a) the crash lands even with NO short: the run crashes honey for everyone. g = _game() g.deck = EventDeck(seed=0, events=[run]) g.tempt_fate() # binds run_on_hoard (honey) g.world.last_price["honey"] = 10.0 # known pre-crash price; no-op policy won't move it g.step() assert g.world.last_price["honey"] == 5.0 # halved at settlement assert any(e["type"] == "legend_crash" and e["good"] == "honey" for e in g.world.log) # (b) a short that front-ran the run settles into profit: (10 - 5) * 4 = +20. h = _game() h.deck = EventDeck(seed=0, events=[run]) h.world.patron.shorts.append(ShortPosition(good="honey", qty=4, entry_price=10.0)) h.tempt_fate() h.world.last_price["honey"] = 10.0 h.step() assert h.world.patron.realized_pnl > 0 # --- (4) crossing the threshold opens an investigation at testimony ------------ def test_heat_threshold_opens_investigation_at_testimony(): g = _game() g.world.patron.heat = INVESTIGATION_THRESHOLD g.step() assert g.world.investigation is not None assert g.world.investigation.stage == "testimony" assert any(e["type"] == "investigation_open" for e in g.world.log) # --- (5) the stage machine: testimony -> evidence -> verdict over 3 turns ------ def test_stage_machine_advances_to_verdict_over_three_turns(): g = _game() g.world.patron.heat = INVESTIGATION_THRESHOLD g.step() # opens at testimony assert g.world.investigation.stage == "testimony" g.step() # advances to evidence assert g.world.investigation.stage == "evidence" g.step() # advances to verdict and closes the case assert g.world.investigation is None assert any(e["type"] == "investigation_verdict" for e in g.world.log) # --- (6) each verdict effect: fine, frozen, exile ------------------------------ def _drive_to_verdict(g): """Step the three turns that open, build, and rule on an investigation.""" g.step() g.step() g.step() def test_verdict_fine_docks_one_hundred_pebbles(): g = _game() g.world.patron.heat = INVESTIGATION_THRESHOLD # 70 -> below 80 -> fine purse0 = g.world.patron.purse _drive_to_verdict(g) verdicts = [e for e in g.world.log if e["type"] == "investigation_verdict"] assert verdicts and verdicts[-1]["verdict"] == "fine" assert g.world.patron.purse == purse0 - 100 def test_verdict_frozen_blocks_a_spending_move_next_turn(): g = _game() g.world.patron.heat = 85 # >=80, <95, no big evidence -> frozen _drive_to_verdict(g) verdicts = [e for e in g.world.log if e["type"] == "investigation_verdict"] assert verdicts and verdicts[-1]["verdict"] == "frozen" assert g.world.turn <= g.world.patron.frozen_until # currently frozen # A spending move next turn is blocked with reason "frozen"; noop is not. g.world.pending_moves.append({"kind": "lend", "target": "Mossback", "amount": 50}) target_pebbles0 = g.world.creatures["Mossback"].pebbles g.step() blocked = [e for e in g.world.log if e["type"] == "player_move" and e.get("reason") == "frozen"] assert blocked assert g.world.creatures["Mossback"].pebbles == target_pebbles0 # the lend never paid def test_verdict_exile_removes_creature_from_alive(): g = _game() g.world.patron.heat = 99 # >=95 -> exile _drive_to_verdict(g) verdicts = [e for e in g.world.log if e["type"] == "investigation_verdict"] assert verdicts and verdicts[-1]["verdict"] == "exile" name = verdicts[-1]["creature"] assert name in g.world.exiled assert name not in {c.name for c in g.world.alive()} assert any(e["type"] == "exile" and e["creature"] == name for e in g.world.log) # --- (7) exile does not corrupt clearing; pebbles conserved across the window -- def test_exile_preserves_market_clearing_and_conserves_pebbles(): g = Game(make_random_policy(seed=5), deck_seed=1) g.world.patron.heat = 99 total0 = sum(c.pebbles for c in g.world.creatures.values()) for _ in range(EXILE_CHAPTER_TURNS + 2): g.step() for c in g.world.creatures.values(): assert c.pebbles >= 0 assert all(q >= 0 for q in c.inventory.values()) # Trading only moves pebbles between creatures; production/consumption never # touch them. The exiled creature's pebbles are hidden, not destroyed, so the # FULL-cast total (alive + exiled) is invariant across the window. assert sum(c.pebbles for c in g.world.creatures.values()) == total0 # --- (8) re-entry restores the creature intact and emits reentry --------------- def test_reentry_restores_creature_and_emits_event(): g = _game() g.world.patron.heat = 99 _drive_to_verdict(g) name = [e for e in g.world.log if e["type"] == "exile"][-1]["creature"] snapshot_pebbles = g.world.creatures[name].pebbles snapshot_inv = dict(g.world.creatures[name].inventory) # While exiled the creature is hidden, not destroyed: its state is frozen. # (A no-op policy means even alive creatures do not trade, isolating the test.) g.step() assert name in g.world.exiled # still serving the exile term assert g.world.creatures[name].pebbles == snapshot_pebbles assert g.world.creatures[name].inventory == snapshot_inv # Step until the term expires and the creature returns to the wood. while name in g.world.exiled: g.step() assert name in {c.name for c in g.world.alive()} assert any(e["type"] == "reentry" and e["creature"] == name for e in g.world.log) # --- (9) peek determinism: peek().key == draw().key; repeated peek is stable --- def test_peek_matches_draw_and_is_stable(): from ttw.events import EventDeck from ttw.world import seed_world deck = EventDeck(seed=3) first = deck.peek() assert deck.peek().key == first.key # repeated peek is stable assert deck.peek() is deck.peek() world = seed_world() assert deck.draw(world).key == first.key # the draw honors the peek # And it holds again on the next, freshly-peeked draw. nxt = deck.peek() assert deck.draw(world).key == nxt.key