mirror of
https://github.com/Heretek-AI/AetherTable.git
synced 2026-11-08 06:58:21 -10:00
John SmithandClaude
68a69c44c1
Deterministic per-seed tactical layer: pending-OA disclosures queue opponent reaction attacks (host drives ownerless attackers), 15% seeded ready actions release only when the structured trigger observably fired. Reaction attempts count as valid attempts, never compliance failures; decision_seed=None preserves legacy behavior byte-identically. 20 tests red-first; pytest 989 passed / 29 skipped. Co-Authored-By: Claude <[email protected]>
1096 lines
49 KiB
Python
1096 lines
49 KiB
Python
"""Backlog item 5.1 (+5.2): agent-driven campaign simulation harness.
|
|
|
|
Contract under test:
|
|
|
|
- ``CampaignSimulation.run()`` drives N synthetic ``CampaignSimPlayer``s
|
|
through the REAL orchestrator surface: auth signup/login, lobby
|
|
create/join/launch, character create/deploy, then T rounds of turns whose
|
|
actions execute through the SAME authenticated proxy endpoints a browser
|
|
uses (``?token=`` forwarded identity).
|
|
- Every LLM decision goes through ``routing.llm_client.LLMStreamingGateway``
|
|
(so JSONL logging captures it); malformed/absent LLM output falls back to
|
|
the deterministic scripted policy FOR THAT TURN while still counting the
|
|
attempt.
|
|
- With no LLM key configured the run is tagged ``mode: "scripted"`` and never
|
|
touches the network.
|
|
- Rejections reported by the proxy/engine are accounted per-round with their
|
|
reasons. The report invents nothing: every number is counted.
|
|
|
|
No unit test touches the network: outbound orchestrator HTTP rides an
|
|
in-process ASGI transport, the authoritative Rust engine behind
|
|
``routing.engine_client`` is replaced with an in-memory fake, and the LLM
|
|
upstream is a canned httpx fake (same pattern as test_llm_intent.py).
|
|
"""
|
|
|
|
import asyncio
|
|
import copy
|
|
import json
|
|
import re
|
|
import uuid as uuid_mod
|
|
|
|
import httpx
|
|
import pytest
|
|
from types import SimpleNamespace
|
|
|
|
from vtt_orchestrator.agents.agent_hierarchy import DirectorAgent
|
|
from vtt_orchestrator.routing import engine_client, llm_client as llm_client_module
|
|
from vtt_orchestrator.routing.engine_client import EngineRejectedError
|
|
from vtt_orchestrator.routing.llm_client import LLMConfig, LLMStreamingGateway
|
|
from vtt_orchestrator.agents.npc_sub_agent import ConcordiaNPC, SocialNorm, SocialNormsComponent
|
|
from vtt_orchestrator.server import app
|
|
from vtt_orchestrator.simulation.campaign_sim import (
|
|
CampaignSimulation,
|
|
CampaignSimPlayer,
|
|
scripted_decision,
|
|
scripted_social_decision,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fakes: the authoritative Rust engine (behind routing.engine_client)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
DUMMY_ID = engine_client._coerce_uuid("sim-training-dummy")
|
|
SOCIAL_NPC_ID = str(DUMMY_ID)
|
|
|
|
|
|
def _dist(a, b) -> float:
|
|
"""Euclidean distance between two [x, y, z] positions; inf when either is
|
|
missing so an unpositioned entity is never treated as adjacent."""
|
|
try:
|
|
return ((float(a[0]) - float(b[0])) ** 2 +
|
|
(float(a[1]) - float(b[1])) ** 2) ** 0.5
|
|
except (TypeError, ValueError, IndexError):
|
|
return float("inf")
|
|
|
|
|
|
class FakeEngine:
|
|
"""In-memory stand-in for vtt-server's HTTP surface.
|
|
|
|
Records every call INCLUDING the forwarded actor identity so tests can
|
|
assert that player tokens (not the gateway service principal) authorized
|
|
each action.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.calls = []
|
|
self.entities = {} # session_id -> {entity_id: entity dict}
|
|
self.reject_rules = [] # list of (path_substring, status, reason)
|
|
self.attack_damage = 3
|
|
# Iteration 81: entities with an ARMED opportunity-attack reaction.
|
|
# When one of these loses adjacency to a mover via /move, the fake
|
|
# mirrors vtt-core's move_entity disclosure: it PENDS the swing
|
|
# (available: true, pending_opportunity naming the endpoint that
|
|
# takes it) instead of resolving silently.
|
|
self.armed_reactions = set()
|
|
self.pending_oa: list = [] # (attacker_id, mover_id) pairs
|
|
self._last_session = None
|
|
|
|
# -- configuration ------------------------------------------------------
|
|
|
|
def reject(self, path_substring: str, reason: str, status: int = 409):
|
|
self.reject_rules.append((path_substring, status, reason))
|
|
|
|
# -- fake transport -----------------------------------------------------
|
|
|
|
async def engine_request(self, method, path, payload=None, *, actor=None):
|
|
self.calls.append(
|
|
{"method": method, "path": path, "payload": payload, "actor": actor,
|
|
"_seq": len(self.calls)}
|
|
)
|
|
m0 = re.fullmatch(r"/api/v1/sessions/([^/]+)", path)
|
|
if method == "GET" and m0:
|
|
self._last_session = m0.group(1)
|
|
elif "/api/v1/sessions/" in path:
|
|
parts = path.split("/")
|
|
if len(parts) > 4:
|
|
self._last_session = parts[4]
|
|
for pattern, status, reason in self.reject_rules:
|
|
if pattern in path:
|
|
raise EngineRejectedError(status, json.dumps({"reason": reason}))
|
|
|
|
if method == "POST" and path == "/api/v1/sessions":
|
|
session_id = str(uuid_mod.uuid4())
|
|
self.entities[session_id] = {}
|
|
return {"session_id": session_id, "status": "CREATED"}
|
|
|
|
m = re.fullmatch(r"/api/v1/sessions/([^/]+)/entities", path)
|
|
if m:
|
|
session_id = m.group(1)
|
|
bucket = self.entities.setdefault(session_id, {})
|
|
entity = dict(payload)
|
|
bucket[entity["id"]] = entity
|
|
return {"status": "SPAWNED", "entity_id": entity["id"]}
|
|
|
|
m = re.fullmatch(r"/api/v1/sessions/([^/]+)", path)
|
|
if method == "GET" and m:
|
|
return {
|
|
"session_id": m.group(1),
|
|
"entities": copy.deepcopy(self.entities.get(m.group(1), {})),
|
|
}
|
|
|
|
m = re.fullmatch(r"/api/v1/sessions/([^/]+)/action/attack", path)
|
|
if m:
|
|
attacker_id = payload["attacker_id"]
|
|
target_id = payload["target_id"]
|
|
bucket = self.entities[m.group(1)]
|
|
if attacker_id not in bucket:
|
|
raise EngineRejectedError(404, json.dumps({"reason": "UNKNOWN_ATTACKER"}))
|
|
target = bucket.get(target_id)
|
|
if target is None or target.get("is_dead"):
|
|
raise EngineRejectedError(409, json.dumps({"reason": "TARGET_ALREADY_DEAD"}))
|
|
target["current_hp"] = int(target.get("current_hp", 0)) - self.attack_damage
|
|
dead = target["current_hp"] <= 0
|
|
target["current_hp"] = max(target["current_hp"], 0)
|
|
target["is_dead"] = dead
|
|
return {
|
|
"status": "RESOLVED",
|
|
"action_name": "Strike",
|
|
"is_hit": True,
|
|
"total_damage": self.attack_damage,
|
|
"target_hp_remaining": target["current_hp"],
|
|
"target_is_dead": dead,
|
|
"event_sequence": len(self.calls),
|
|
}
|
|
|
|
m = re.fullmatch(r"/api/v1/sessions/([^/]+)/move", path)
|
|
if m:
|
|
bucket = self.entities[m.group(1)]
|
|
entity = bucket.get(payload["entity_id"])
|
|
if entity is None:
|
|
raise EngineRejectedError(409, json.dumps({"reason": "MOVE_REJECTED"}))
|
|
mover_id = str(entity.get("id"))
|
|
was_adjacent_to = [
|
|
other_id for other_id, other in bucket.items()
|
|
if other_id != mover_id
|
|
and not other.get("is_dead")
|
|
and bool(other.get("is_player")) != bool(entity.get("is_player"))
|
|
and _dist(entity.get("position"), other.get("position")) <= 5.0
|
|
]
|
|
entity["position"] = [payload["x"], payload["y"], 0.0]
|
|
provoked = [
|
|
attacker for attacker in was_adjacent_to
|
|
if attacker in self.armed_reactions
|
|
and _dist(bucket[attacker].get("position"), entity.get("position")) > 5.5
|
|
]
|
|
outcome = {"to": {"x": payload["x"], "y": payload["y"]}}
|
|
body: dict = {"status": "MOVED", "outcome": outcome}
|
|
if provoked:
|
|
detail = []
|
|
for attacker in provoked:
|
|
self.pending_oa.append((attacker, mover_id))
|
|
detail.append({
|
|
"provoked_by": attacker,
|
|
"reaction_type": "opportunity_attack",
|
|
"pending_opportunity": "/action/opportunity-attack",
|
|
"available": True,
|
|
})
|
|
body["opportunity_attacks_detail"] = detail
|
|
return body
|
|
|
|
if path == "/api/v1/actions/check":
|
|
return {"total": 15, "d20": 13, "modifier": 2,
|
|
"dc": payload.get("dc"), "success": True}
|
|
|
|
if path.endswith("/action/opportunity-attack"):
|
|
pair = (payload["attacker_id"], payload["target_id"])
|
|
if pair not in self.pending_oa:
|
|
return {
|
|
"ok": False,
|
|
"error": "OA_NOT_PENDING",
|
|
"message": "no pending opportunity attack against this mover",
|
|
}
|
|
self.pending_oa.remove(pair)
|
|
attacker = self.entities[self._last_session][pair[0]]
|
|
attacker["action_budget"]["reaction"] = False
|
|
target = self.entities[self._last_session][pair[1]]
|
|
target["current_hp"] = max(0, int(target.get("current_hp", 0)) - 2)
|
|
return {"status": "OPPORTUNITY_ATTACK_TAKEN", "is_hit": True,
|
|
"total_damage": 2, "reaction_spent": True}
|
|
|
|
if path.endswith("/action/ready"):
|
|
entity = self.entities[self._last_session][payload["entity_id"]]
|
|
entity["readied_action"] = {
|
|
"description": payload["description"],
|
|
"trigger": payload.get("trigger") or "",
|
|
}
|
|
entity["action_budget"]["action"] = False
|
|
return {"status": "READY_ACTION_SET",
|
|
"entity_id": payload["entity_id"],
|
|
"readied_action": entity["readied_action"]}
|
|
|
|
if path.endswith("/action/ready/release"):
|
|
entity = self.entities[self._last_session].get(payload["entity_id"])
|
|
readied = (entity or {}).pop("readied_action", None)
|
|
if readied is None:
|
|
return {"ok": False, "error": "NO_READIED_ACTION",
|
|
"message": "nothing is readied"}
|
|
entity["action_budget"]["reaction"] = False
|
|
return {"status": "READY_ACTION_RELEASED",
|
|
"entity_id": payload["entity_id"],
|
|
"released_action": readied, "reaction_spent": True}
|
|
|
|
if path == "/api/v1/actions/check":
|
|
return {"total": 15, "d20": 13, "modifier": 2,
|
|
"dc": payload.get("dc"), "success": True}
|
|
|
|
raise AssertionError(f"FakeEngine has no route for {method} {path}")
|
|
|
|
def calls_to(self, substring):
|
|
return [c for c in self.calls if substring in c["path"]]
|
|
|
|
|
|
@pytest.fixture()
|
|
def fake_engine(monkeypatch):
|
|
"""Replaces the engine transport AND clears ambient LLM credentials so a
|
|
developer's exported LLM_KEY can never turn a scripted test nondeterministic."""
|
|
fake = FakeEngine()
|
|
monkeypatch.setattr(engine_client, "engine_request", fake.engine_request)
|
|
for var in ("LLM_KEY", "OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GEMINI_API_KEY"):
|
|
monkeypatch.delenv(var, raising=False)
|
|
monkeypatch.delenv("OLLAMA_BASE_URL", raising=False)
|
|
return fake
|
|
|
|
|
|
def asgi_transport():
|
|
return httpx.ASGITransport(app=app)
|
|
|
|
|
|
def make_simulation(players=2, rounds=3, **kwargs):
|
|
kwargs.setdefault("transport", asgi_transport())
|
|
return CampaignSimulation(players=players, rounds=rounds, **kwargs)
|
|
|
|
|
|
def run(players=2, rounds=3, **kwargs):
|
|
return asyncio.run(make_simulation(players, rounds, **kwargs).run())
|
|
|
|
|
|
def run_with_sim(players=2, rounds=3, **kwargs):
|
|
"""Like ``run`` but also hands back the simulation instance so tests can
|
|
inspect its disposition engine directly."""
|
|
sim = make_simulation(players, rounds, **kwargs)
|
|
return sim, asyncio.run(sim.run())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fakes: the LLM upstream (canned chat-completions bodies)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _install_upstream(monkeypatch, contents):
|
|
"""Patch ONLY the LLM leg of httpx: chat-completions posts get canned
|
|
bodies, every other URL passes through to the real client (which is how
|
|
the sim's own ASGI transport keeps working)."""
|
|
bodies = [
|
|
c if isinstance(c, dict) else {
|
|
"choices": [{"message": {"role": "assistant", "content": c}}]
|
|
}
|
|
for c in contents
|
|
]
|
|
|
|
class _FakeAsyncClient(httpx.AsyncClient):
|
|
remaining = list(bodies)
|
|
calls = []
|
|
|
|
async def post(self, url, headers=None, json=None, **kwargs):
|
|
if "chat/completions" not in str(url):
|
|
return await super().post(url, headers=headers, json=json, **kwargs)
|
|
type(self).calls.append({"url": url, "payload": json})
|
|
body = type(self).remaining.pop(0)
|
|
return httpx.Response(200, json=body,
|
|
request=httpx.Request("POST", str(url)))
|
|
|
|
monkeypatch.setattr(llm_client_module.httpx, "AsyncClient", _FakeAsyncClient)
|
|
return _FakeAsyncClient
|
|
|
|
|
|
@pytest.fixture()
|
|
def llm_mode(monkeypatch, tmp_path):
|
|
# Keep run artifacts out of the shared logs/ directory during tests.
|
|
monkeypatch.setattr(llm_client_module, "LLM_LOG_PATH", str(tmp_path / "llm_calls.jsonl"))
|
|
monkeypatch.setenv("LLM_API", "http://fake-llm.test/v1")
|
|
monkeypatch.setenv("LLM_KEY", "test-key")
|
|
monkeypatch.setenv("LLM_MODEL", "test-model")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scripted-mode happy path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestScriptedModeLifecycle:
|
|
def test_full_lifecycle_end_to_end_against_mocks(self, fake_engine):
|
|
report = run(players=2, rounds=3)
|
|
|
|
assert report["mode"] == "scripted"
|
|
assert report["players_configured"] == 2
|
|
assert report["lobby_id"]
|
|
assert report["engine_session_id"] in fake_engine.entities
|
|
|
|
# Lobby lifecycle really happened through the orchestrator.
|
|
assert len(fake_engine.calls_to("/entities")) >= 3 # dummy + 2 deploys
|
|
assert any(c["path"] == "/api/v1/sessions" and c["method"] == "POST"
|
|
for c in fake_engine.calls)
|
|
|
|
totals = report["totals"]
|
|
assert totals["actions_attempted"] == 6 # 2 players x 3 rounds
|
|
assert totals["accepted"] + totals["rejected"] == totals["actions_attempted"]
|
|
assert totals["llm_calls_made"] == 0 # scripted mode never asks
|
|
assert all(t["decision_source"] == "scripted" for r in report["rounds"]
|
|
for t in r["turns"])
|
|
assert len(report["rounds"]) == 3
|
|
|
|
def test_every_proxied_action_forwards_the_players_own_identity(self, fake_engine):
|
|
report = run(players=2, rounds=3)
|
|
|
|
user_ids = {p["user_id"] for p in report["per_player"]}
|
|
assert len(user_ids) == 2
|
|
action_calls = [c for c in fake_engine.calls
|
|
if any(s in c["path"] for s in ("/move", "/attack"))]
|
|
assert action_calls, "expected at least one proxied action"
|
|
for call in action_calls:
|
|
assert call["actor"] is not None, "action proxy must forward caller identity"
|
|
assert call["actor"]["user_id"] in user_ids
|
|
|
|
def test_deploy_binds_entities_and_dummy_is_hostile_target(self, fake_engine):
|
|
report = run(players=2, rounds=3)
|
|
session_state = fake_engine.entities[report["engine_session_id"]]
|
|
pc_ids = {p["entity_id"] for p in report["per_player"]}
|
|
assert len(pc_ids) == 2
|
|
for entity_id, entity in session_state.items():
|
|
if entity_id in pc_ids:
|
|
assert entity["owner_player_id"] in {
|
|
p["user_id"] for p in report["per_player"]}
|
|
else:
|
|
assert entity.get("is_player") is False # the training dummy
|
|
|
|
attacks = fake_engine.calls_to("/action/attack")
|
|
assert attacks, "scripted policy must attack the hostile dummy"
|
|
assert all(c["payload"]["target_id"] == DUMMY_ID for c in attacks)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rejection accounting
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRejectionAccounting:
|
|
def test_engine_rejections_are_counted_with_reasons(self, fake_engine):
|
|
fake_engine.reject("/move", "MOVE_REJECTED")
|
|
|
|
report = run(players=2, rounds=2)
|
|
|
|
totals = report["totals"]
|
|
assert totals["rejected"] > 0
|
|
assert totals["rejected"] == totals["actions_attempted"] - totals["accepted"]
|
|
assert totals["rejection_reasons"].get("MOVE_REJECTED") == totals["rejected"]
|
|
# Attempted still counts the refused turns — nothing disappears.
|
|
assert totals["actions_attempted"] == 4
|
|
rejected_turns = [t for r in report["rounds"] for t in r["turns"] if t["rejected"]]
|
|
assert len(rejected_turns) == totals["rejected"]
|
|
assert all(t["rejection_reason"] == "MOVE_REJECTED" for t in rejected_turns)
|
|
|
|
def test_partial_rejection_mix_is_accounted_per_round(self, fake_engine):
|
|
fake_engine.reject("/action/attack", "TARGET_OUT_OF_RANGE")
|
|
|
|
report = run(players=2, rounds=3)
|
|
assert report["totals"]["rejection_reasons"] == {"TARGET_OUT_OF_RANGE":
|
|
report["totals"]["rejected"]}
|
|
per_round_attempts = sum(len(r["turns"]) for r in report["rounds"])
|
|
assert per_round_attempts == report["totals"]["actions_attempted"] == 6
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLM mode
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestLLMMode:
|
|
def test_valid_llm_decisions_are_parsed_and_executed(self, fake_engine, llm_mode, monkeypatch):
|
|
upstream = _install_upstream(monkeypatch, [
|
|
json.dumps({"action": "attack", "target_id": DUMMY_ID, "reason": "strike it"}),
|
|
json.dumps({"action": "move", "x": 7.5, "y": 6.0}),
|
|
json.dumps({"action": "check"}),
|
|
json.dumps({"action": "attack", "target_id": DUMMY_ID}),
|
|
])
|
|
|
|
report = run(players=2, rounds=2)
|
|
|
|
assert report["mode"] == "llm"
|
|
assert len(upstream.calls) == 4 # one per turn
|
|
assert all("chat/completions" in c["url"] for c in upstream.calls)
|
|
totals = report["totals"]
|
|
assert totals["llm_calls_made"] == 4
|
|
assert totals["llm_decisions_accepted"] == 4
|
|
assert totals["llm_fallbacks"] == 0
|
|
sources = [t["decision_source"] for r in report["rounds"] for t in r["turns"]]
|
|
assert sources.count("llm") == 4
|
|
# The attack decisions really reached the engine against the named target.
|
|
attacks = fake_engine.calls_to("/action/attack")
|
|
assert len(attacks) == 2
|
|
assert all(c["payload"]["target_id"] == DUMMY_ID for c in attacks)
|
|
assert len(fake_engine.calls_to("/move")) == 1
|
|
|
|
def test_malformed_llm_output_falls_back_but_counts_the_attempt(self, fake_engine, llm_mode, monkeypatch):
|
|
_install_upstream(monkeypatch, ["this is not json {{{"])
|
|
|
|
report = run(players=1, rounds=1)
|
|
|
|
assert report["mode"] == "llm"
|
|
totals = report["totals"]
|
|
assert totals["llm_calls_made"] == 1 # attempt counted
|
|
assert totals["llm_fallbacks"] == 1
|
|
assert totals["llm_decisions_accepted"] == 0
|
|
turn = report["rounds"][0]["turns"][0]
|
|
assert turn["decision_source"] == "llm_fallback"
|
|
assert turn["fallback_reason"]
|
|
# The turn STILL happened via the deterministic policy.
|
|
assert totals["actions_attempted"] == 1
|
|
|
|
def test_unknown_action_label_falls_back_to_scripted_choice(self, fake_engine, llm_mode, monkeypatch):
|
|
_install_upstream(monkeypatch, ['{"action": "summon_dragon"}'])
|
|
|
|
report = run(players=1, rounds=1)
|
|
|
|
turn = report["rounds"][0]["turns"][0]
|
|
assert turn["decision_source"] == "llm_fallback"
|
|
assert "unknown_action" in turn["fallback_reason"]
|
|
assert turn["action"] in ("attack", "move", "check")
|
|
|
|
def test_llm_http_failure_falls_back_and_run_stays_honest(self, fake_engine, llm_mode, monkeypatch):
|
|
class _FailingClient(httpx.AsyncClient):
|
|
async def post(self, url, *args, **kwargs):
|
|
if "chat/completions" in str(url):
|
|
raise RuntimeError("connection reset")
|
|
return await super().post(url, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(llm_client_module.httpx, "AsyncClient", _FailingClient)
|
|
|
|
report = run(players=1, rounds=2)
|
|
|
|
assert report["mode"] == "llm" # a key WAS configured
|
|
assert report["totals"]["llm_fallbacks"] == 2
|
|
assert report["totals"]["actions_attempted"] == 2
|
|
|
|
def test_no_key_configured_means_scripted_never_touches_network(self, fake_engine, monkeypatch):
|
|
class _NeverClient(httpx.AsyncClient):
|
|
async def post(self, url, *args, **kwargs):
|
|
if "chat/completions" in str(url):
|
|
raise AssertionError("network LLM call attempted in scripted mode")
|
|
return await super().post(url, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(llm_client_module.httpx, "AsyncClient", _NeverClient)
|
|
|
|
# The social NPC is injected WITH an LLM gateway of its own: scripted
|
|
# mode must pin llm_gateway=None on every respond_to call so the
|
|
# persona's own gateway can never leak an upstream call past the
|
|
# sim's "scripted never touches the network" guarantee.
|
|
class _RecordingReplyGateway:
|
|
"""Records every call instead of raising — respond_to swallows
|
|
gateway exceptions by contract, so raising here would hide the
|
|
very leak this test exists to catch."""
|
|
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
async def complete_json(self, system_prompt, user_prompt):
|
|
self.calls.append((system_prompt, user_prompt))
|
|
return {"reply": "The shrine remembers you."}
|
|
|
|
reply_gateway = _RecordingReplyGateway()
|
|
report = run(
|
|
players=2, rounds=2,
|
|
npc_registry={SOCIAL_NPC_ID: _make_npc(gateway=reply_gateway)},
|
|
social_npc_id=SOCIAL_NPC_ID,
|
|
)
|
|
|
|
assert report["mode"] == "scripted"
|
|
assert report["totals"]["llm_calls_made"] == 0
|
|
# Zero calls through the persona's own gateway — the sim pinned
|
|
# llm_gateway=None for the whole run.
|
|
assert reply_gateway.calls == []
|
|
# Every exchange still happened — answered by the deterministic
|
|
# template voice, never an upstream model.
|
|
socials = [e for r in report["rounds"] for e in r["social"]]
|
|
assert len(socials) == 4 # 2 players x 2 rounds
|
|
assert all(e["reply_generator"] == "template" for e in socials)
|
|
|
|
|
|
class _InstallNever:
|
|
"""Base for failing clients used by TestLLMMode.test_llm_http_failure."""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Player-level auth behavior
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPlayerAuth:
|
|
def test_signup_then_login_for_existing_email(self, fake_engine):
|
|
async def scenario():
|
|
transport = asgi_transport()
|
|
first = CampaignSimPlayer("alpha", 0, transport=transport,
|
|
email="[email protected]")
|
|
await first.authenticate()
|
|
second = CampaignSimPlayer("alpha", 0, transport=transport,
|
|
email="[email protected]")
|
|
await second.authenticate() # same email -> login path, not signup
|
|
assert first.user_id == second.user_id
|
|
# Both tokens actually authorize authenticated calls.
|
|
record = await second.authed("GET", "/api/v1/auth/session")
|
|
assert record["status"] == 200 and record["body"]["valid"] is True
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Report honesty / shape
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestReportShape:
|
|
REQUIRED_KEYS = (
|
|
"mode", "started_at", "elapsed_seconds", "players_configured",
|
|
"rounds_requested", "lobby_id", "engine_session_id",
|
|
"totals", "per_player", "rounds", "errors",
|
|
)
|
|
|
|
TOTALS_KEYS = (
|
|
"turns", "actions_attempted", "accepted", "rejected",
|
|
"rejection_reasons", "llm_calls_made", "llm_decisions_accepted",
|
|
"llm_fallbacks", "degraded_flags_seen",
|
|
)
|
|
|
|
def test_report_contains_only_counted_metrics(self, fake_engine):
|
|
report = run(players=2, rounds=2)
|
|
|
|
for key in self.REQUIRED_KEYS:
|
|
assert key in report, f"missing report key {key}"
|
|
for key in self.TOTALS_KEYS:
|
|
assert key in report["totals"], f"missing totals key {key}"
|
|
|
|
totals = report["totals"]
|
|
assert totals["turns"] == 4
|
|
assert totals["accepted"] + totals["rejected"] == totals["actions_attempted"]
|
|
assert sum(totals["rejection_reasons"].values()) == totals["rejected"]
|
|
|
|
# No fabricated benchmark scores: MCR/HCI are the playtest harness's
|
|
# job, not this report's.
|
|
assert "mechanical_compliance_rate_pct" not in report
|
|
assert "hallucination_continuity_index" not in report
|
|
|
|
assert len(report["per_player"]) == 2
|
|
for entry in report["per_player"]:
|
|
assert entry["actions_attempted"] == 2
|
|
assert entry["accepted"] + entry["rejected"] == 2
|
|
assert entry["entity_id"] and entry["character_id"] and entry["user_id"]
|
|
assert sum(p["actions_attempted"] for p in report["per_player"]) == \
|
|
totals["actions_attempted"]
|
|
|
|
def test_degraded_flags_seen_counts_truthy_degraded_fields(self, fake_engine, monkeypatch):
|
|
original = fake_engine.engine_request
|
|
|
|
async def tagging(method, path, payload=None, *, actor=None):
|
|
result = await original(method, path, payload, actor=actor)
|
|
if isinstance(result, dict) and "/move" in path:
|
|
result["degraded"] = True
|
|
return result
|
|
|
|
monkeypatch.setattr(engine_client, "engine_request", tagging)
|
|
report = run(players=1, rounds=2)
|
|
assert report["totals"]["degraded_flags_seen"] > 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Social-state telemetry (campaign_sim x npc_disposition integration)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _all_turns(report):
|
|
return [t for r in report["rounds"] for t in r["turns"]]
|
|
|
|
|
|
class TestSocialStateTelemetry:
|
|
def test_accepted_attacks_record_attacked_kind_on_target(self, fake_engine):
|
|
sim, report = run_with_sim(players=2, rounds=3)
|
|
|
|
history = sim.disposition.history()
|
|
accepted_attacks = [t for t in _all_turns(report)
|
|
if t["action"] == "attack" and t["accepted"]]
|
|
assert accepted_attacks, "scripted policy must attack in round 1"
|
|
assert len(history) == len(accepted_attacks)
|
|
# Exactly the expected directed kind onto the attacked target.
|
|
assert all(rec.kind == "attacked" for rec in history)
|
|
assert all(rec.npc_id == DUMMY_ID for rec in history)
|
|
entity_ids = {p["entity_id"] for p in report["per_player"]}
|
|
assert all(rec.player_id in entity_ids for rec in history)
|
|
|
|
def test_checks_and_moves_record_no_disposition_entries(self, fake_engine):
|
|
sim, report = run_with_sim(players=2, rounds=4)
|
|
|
|
non_attack_turns = [t for t in _all_turns(report) if t["action"] != "attack"]
|
|
assert any(t["accepted"] for t in non_attack_turns), \
|
|
"scenario must contain accepted checks/moves"
|
|
index_of = {p["name"]: i for i, p in enumerate(report["per_player"])}
|
|
attack_ts = {t["round"] * 10.0 + index_of[t["player"]]
|
|
for t in _all_turns(report) if t["action"] == "attack"}
|
|
recorded_ts = {rec.timestamp for rec in sim.disposition.history()}
|
|
# Every recorded interaction sits on an attack turn's sim-clock slot —
|
|
# checks/moves contributed nothing.
|
|
assert recorded_ts <= attack_ts
|
|
assert all(rec.kind == "attacked" for rec in sim.disposition.history())
|
|
|
|
def test_rejected_attacks_record_nothing(self, fake_engine):
|
|
fake_engine.reject("/action/attack", "TARGET_OUT_OF_RANGE")
|
|
|
|
sim, report = run_with_sim(players=2, rounds=2)
|
|
|
|
attacks = [t for t in _all_turns(report) if t["action"] == "attack"]
|
|
assert attacks and all(t["rejected"] for t in attacks)
|
|
assert sim.disposition.history() == []
|
|
assert report["totals"]["disposition_interactions"] == 0
|
|
|
|
def test_report_carries_per_player_stances_and_interaction_totals(self, fake_engine):
|
|
sim, report = run_with_sim(players=2, rounds=2)
|
|
|
|
totals = report["totals"]
|
|
history = sim.disposition.history()
|
|
assert totals["disposition_interactions"] == len(history) > 0
|
|
assert totals["disposition_interactions_by_kind"] == {
|
|
"attacked": len(history),
|
|
}
|
|
for entry in report["per_player"]:
|
|
# One accepted attack each -> the dummy tracks a stance toward them,
|
|
# keyed by display name (name preferred over raw id).
|
|
assert entry["interactions_recorded"] == 1
|
|
assert entry["stances"] == {
|
|
"Training Dummy":
|
|
sim.disposition.stance(DUMMY_ID, entry["entity_id"]),
|
|
}
|
|
|
|
def test_hostile_stance_after_repeated_attacks(self, fake_engine, llm_mode, monkeypatch):
|
|
_install_upstream(monkeypatch, [
|
|
json.dumps({"action": "attack", "target_id": DUMMY_ID})
|
|
] * 3)
|
|
|
|
sim, report = run_with_sim(players=1, rounds=3)
|
|
|
|
kinds = [rec.kind for rec in sim.disposition.history()]
|
|
assert kinds == ["attacked", "attacked", "attacked"]
|
|
player = report["per_player"][0]
|
|
assert player["interactions_recorded"] == 3
|
|
assert player["stances"] == {"Training Dummy": "hostile"}
|
|
|
|
def test_decision_prompt_includes_current_target_stances(self, fake_engine,
|
|
llm_mode, monkeypatch):
|
|
upstream = _install_upstream(monkeypatch, [
|
|
json.dumps({"action": "check"}),
|
|
])
|
|
|
|
run_with_sim(players=1, rounds=1)
|
|
|
|
prompt_text = json.dumps(upstream.calls[0]["payload"]["messages"])
|
|
assert "stances_toward_you" in prompt_text
|
|
# The training dummy has no relationship yet -> surfaced as neutral.
|
|
assert "neutral" in prompt_text
|
|
|
|
def test_scripted_policy_inputs_receive_stances_unknown_targets_neutral(
|
|
self, fake_engine):
|
|
sim = CampaignSimulation(mode="scripted")
|
|
stub = SimpleNamespace(entity_id="pc-1")
|
|
snapshot = {"entities": [
|
|
{"id": "pc-1", "name": "Me", "hp": 28, "max_hp": 28, "ac": 16,
|
|
"is_player": True, "is_dead": False},
|
|
{"id": "npc-9", "name": "Goblin", "hp": 7, "max_hp": 7, "ac": 13,
|
|
"is_player": False, "is_dead": False},
|
|
]}
|
|
|
|
view = sim._stance_view(stub, snapshot)
|
|
assert view == {"Goblin": "neutral"} # unknown target -> neutral
|
|
assert sim._stance_toward("totally-unknown", "pc-1") == "neutral"
|
|
assert sim._stance_toward("x", None) == "neutral"
|
|
|
|
decision = scripted_decision(stub, snapshot, 1, view)
|
|
assert decision["action"] == "attack"
|
|
assert decision["target_id"] == "npc-9"
|
|
|
|
def test_timestamps_deterministic_across_identical_runs(self, fake_engine):
|
|
sim_a, rep_a = run_with_sim(players=2, rounds=3)
|
|
sim_b, rep_b = run_with_sim(players=2, rounds=3)
|
|
|
|
# Each run signs up fresh users, so raw player UUIDs differ; project
|
|
# them onto stable per-player slots before comparing.
|
|
def normalized(report, sim):
|
|
slot = {p["entity_id"]: i for i, p in enumerate(report["per_player"])}
|
|
return [(slot.get(r.player_id, r.player_id), r.npc_id, r.kind,
|
|
r.magnitude, r.timestamp)
|
|
for r in sim.disposition.history()]
|
|
|
|
hist_a = normalized(rep_a, sim_a)
|
|
hist_b = normalized(rep_b, sim_b)
|
|
assert hist_a and hist_a == hist_b
|
|
assert [p["stances"] for p in rep_a["per_player"]] == \
|
|
[p["stances"] for p in rep_b["per_player"]]
|
|
assert [p["interactions_recorded"] for p in rep_a["per_player"]] == \
|
|
[p["interactions_recorded"] for p in rep_b["per_player"]]
|
|
|
|
# Timestamps are pure sim-clock values (round * 10 + turn index),
|
|
# never wall-clock reads.
|
|
index_of = {p["name"]: i for i, p in enumerate(rep_a["per_player"])}
|
|
expected_ts = {t["round"] * 10.0 + index_of[t["player"]]
|
|
for t in _all_turns(rep_a)
|
|
if t["action"] == "attack" and t["accepted"]}
|
|
assert {rec.timestamp for rec in sim_a.disposition.history()} == expected_ts
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Social dialogue phase (campaign_sim x ConcordiaNPC sub-agents)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class _StubReplyGateway:
|
|
"""Canned NPC-side LLM: returns one fixed {"reply": ...} JSON body."""
|
|
|
|
def __init__(self, reply):
|
|
self.reply = reply
|
|
self.calls = []
|
|
|
|
async def complete_json(self, system_prompt, user_prompt):
|
|
self.calls.append((system_prompt, user_prompt))
|
|
return {"reply": self.reply}
|
|
|
|
|
|
def _make_npc(npc_id=SOCIAL_NPC_ID, name="Marrow", gateway=None, norms=None):
|
|
return ConcordiaNPC(
|
|
npc_id=npc_id, name=name, role="Keeper of the Sunken Shrine",
|
|
norms=norms if norms is not None else SocialNormsComponent(),
|
|
llm_gateway=gateway,
|
|
)
|
|
|
|
|
|
def _all_social(report):
|
|
return [e for r in report["rounds"] for e in r["social"]]
|
|
|
|
|
|
class TestSocialDialoguePhase:
|
|
def test_mocked_llm_social_exchange_is_recorded(self, fake_engine, llm_mode, monkeypatch):
|
|
upstream = _install_upstream(monkeypatch, [
|
|
json.dumps({"action": "check"}), # combat turn
|
|
json.dumps({"approach": "aided",
|
|
"utterance": "Hold still, let me dress that wound."}), # social decision
|
|
])
|
|
reply_gateway = _StubReplyGateway("You are kind, stranger.")
|
|
sim, report = run_with_sim(
|
|
players=1, rounds=1,
|
|
npc_registry={SOCIAL_NPC_ID: _make_npc(gateway=reply_gateway)},
|
|
social_npc_id=SOCIAL_NPC_ID,
|
|
)
|
|
|
|
entries = _all_social(report)
|
|
assert len(entries) == 1
|
|
entry = entries[0]
|
|
assert entry["player"] == report["per_player"][0]["name"]
|
|
assert entry["npc_id"] == SOCIAL_NPC_ID
|
|
assert entry["approach"] == "aided"
|
|
assert entry["decision_source"] == "llm"
|
|
assert entry["reply_generator"] == "llm"
|
|
assert entry["reply"] == "You are kind, stranger."
|
|
assert entry["stance_before"] == "neutral"
|
|
# aided -> positive disposition record; +8 trust is still the neutral band.
|
|
assert entry["stance_after"] == "neutral"
|
|
assert "norm_rejected" not in entry
|
|
|
|
# The player decision rode the SAME gateway path as combat decisions.
|
|
assert len(upstream.calls) == 2
|
|
# The reply really came from ConcordiaNPC.respond_to via its own gateway.
|
|
assert len(reply_gateway.calls) == 1
|
|
|
|
kinds = [rec.kind for rec in sim.disposition.history()]
|
|
assert kinds == ["aided"]
|
|
assert report["totals"]["social_interactions"] == 1
|
|
assert report["totals"]["disposition_interactions_by_kind"] == {"aided": 1}
|
|
|
|
def test_norms_violating_llm_reply_degrades_to_template(self, fake_engine, llm_mode, monkeypatch):
|
|
_install_upstream(monkeypatch, [
|
|
json.dumps({"action": "check"}),
|
|
json.dumps({"approach": "gifted", "utterance": "A gift, keeper."}),
|
|
])
|
|
norms = SocialNormsComponent([
|
|
SocialNorm.taboo(["drowned sigil"], reason="never reveal cult secrets"),
|
|
])
|
|
reply_gateway = _StubReplyGateway("The drowned sigil lies below the chapel.")
|
|
sim, report = run_with_sim(
|
|
players=1, rounds=1,
|
|
npc_registry={SOCIAL_NPC_ID: _make_npc(gateway=reply_gateway, norms=norms)},
|
|
social_npc_id=SOCIAL_NPC_ID,
|
|
)
|
|
|
|
entry = _all_social(report)[0]
|
|
# The LLM reply was attempted, rejected by the norms component, and the
|
|
# exchange degraded to the deterministic template voice.
|
|
assert len(reply_gateway.calls) == 1
|
|
assert entry["reply_generator"] == "template"
|
|
assert entry["norm_rejected"] == "never reveal cult secrets"
|
|
assert "drowned sigil" not in entry["reply"]
|
|
assert report["totals"]["social_interactions"] == 1
|
|
|
|
def test_norms_violating_llm_utterance_never_reaches_the_npc_llm(
|
|
self, fake_engine, llm_mode, monkeypatch):
|
|
_install_upstream(monkeypatch, [
|
|
json.dumps({"action": "check"}),
|
|
json.dumps({"approach": "threatened",
|
|
"utterance": "Tell me about the drowned sigil or else."}),
|
|
])
|
|
norms = SocialNormsComponent([
|
|
SocialNorm.taboo(["drowned sigil"], reason="never reveal cult secrets"),
|
|
])
|
|
reply_gateway = _StubReplyGateway("It is buried. I would never say more.")
|
|
sim, report = run_with_sim(
|
|
players=1, rounds=1,
|
|
npc_registry={SOCIAL_NPC_ID: _make_npc(gateway=reply_gateway, norms=norms)},
|
|
social_npc_id=SOCIAL_NPC_ID,
|
|
)
|
|
|
|
entry = _all_social(report)[0]
|
|
assert entry["decision_source"] == "llm"
|
|
# Absolute enforcement: the tainted utterance is swapped for the scripted
|
|
# line AND the NPC's reply is forced down the template path.
|
|
assert len(reply_gateway.calls) == 0
|
|
assert entry["reply_generator"] == "template"
|
|
assert entry["norm_rejected"] == "never reveal cult secrets"
|
|
|
|
def test_hostile_approaches_shift_stance_and_replies_across_rounds(
|
|
self, fake_engine, llm_mode, monkeypatch):
|
|
upstream = _install_upstream(monkeypatch, [
|
|
json.dumps({"action": "check"}),
|
|
json.dumps({"approach": "threatened", "utterance": "Talk, or else."}),
|
|
] * 3)
|
|
sim, report = run_with_sim(
|
|
players=1, rounds=3,
|
|
npc_registry={SOCIAL_NPC_ID: _make_npc()}, # no gateway -> template voice
|
|
social_npc_id=SOCIAL_NPC_ID,
|
|
)
|
|
|
|
entries = _all_social(report)
|
|
assert len(entries) == 3
|
|
assert all(e["decision_source"] == "llm" for e in entries)
|
|
assert all(e["approach"] == "threatened" for e in entries)
|
|
stances = [e["stance_after"] for e in entries]
|
|
assert stances == ["unfriendly", "unfriendly", "hostile"]
|
|
assert entries[0]["stance_before"] == "neutral"
|
|
# Replies must reflect the shifted stance (template cores are per-stance).
|
|
assert "I owe you nothing." in entries[-1]["reply"]
|
|
assert [rec.kind for rec in sim.disposition.history()] == \
|
|
["threatened", "threatened", "threatened"]
|
|
|
|
def test_registry_none_skips_social_phase_entirely(self, fake_engine, llm_mode, monkeypatch):
|
|
upstream = _install_upstream(monkeypatch, [
|
|
json.dumps({"action": "check"}),
|
|
] * 2)
|
|
|
|
sim, report = run_with_sim(players=1, rounds=2)
|
|
|
|
assert all(r["social"] == [] for r in report["rounds"])
|
|
assert report["totals"]["social_interactions"] == 0
|
|
assert sim.disposition.history() == []
|
|
# No extra LLM calls were burned on dialogue that cannot happen.
|
|
assert len(upstream.calls) == 2
|
|
assert all("chat/completions" in c["url"] for c in upstream.calls)
|
|
|
|
def test_designated_npc_absent_from_session_records_nothing(self, fake_engine):
|
|
sim, report = run_with_sim(
|
|
players=1, rounds=2,
|
|
npc_registry={"ghost-npc-id": _make_npc(npc_id="ghost-npc-id")},
|
|
social_npc_id="ghost-npc-id",
|
|
)
|
|
|
|
assert all(r["social"] == [] for r in report["rounds"])
|
|
assert report["totals"]["social_interactions"] == 0
|
|
# Combat telemetry on the training dummy may exist, but the designated
|
|
# (absent) social NPC was never touched.
|
|
assert all(rec.npc_id != "ghost-npc-id" for rec in sim.disposition.history())
|
|
|
|
def test_invalid_llm_approach_falls_back_but_interaction_still_recorded(
|
|
self, fake_engine, llm_mode, monkeypatch):
|
|
_install_upstream(monkeypatch, [
|
|
json.dumps({"action": "check"}),
|
|
json.dumps({"approach": "summon_dragon", "utterance": "hi"}),
|
|
])
|
|
sim, report = run_with_sim(
|
|
players=1, rounds=1,
|
|
npc_registry={SOCIAL_NPC_ID: _make_npc()},
|
|
social_npc_id=SOCIAL_NPC_ID,
|
|
)
|
|
|
|
entry = _all_social(report)[0]
|
|
assert entry["decision_source"] == "llm_fallback"
|
|
assert "unknown_approach" in entry["fallback_reason"]
|
|
# Deterministic scripted fallback for a neutral stance is "gifted".
|
|
assert entry["approach"] == "gifted"
|
|
assert report["totals"]["social_interactions"] == 1
|
|
assert [rec.kind for rec in sim.disposition.history()] == ["gifted"]
|
|
|
|
def test_scripted_mode_records_social_exchanges_without_the_network(
|
|
self, fake_engine, monkeypatch):
|
|
class _NeverClient(httpx.AsyncClient):
|
|
async def post(self, url, *args, **kwargs):
|
|
if "chat/completions" in str(url):
|
|
raise AssertionError("network LLM call attempted in scripted mode")
|
|
return await super().post(url, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(llm_client_module.httpx, "AsyncClient", _NeverClient)
|
|
|
|
sim, report = run_with_sim(
|
|
players=1, rounds=2,
|
|
npc_registry={SOCIAL_NPC_ID: _make_npc()},
|
|
social_npc_id=SOCIAL_NPC_ID,
|
|
)
|
|
|
|
entries = _all_social(report)
|
|
assert len(entries) == 2
|
|
assert all(e["decision_source"] == "scripted" for e in entries)
|
|
assert all(e["reply_generator"] == "template" for e in entries)
|
|
# The scripted approach must follow the CURRENT stance it was handed
|
|
# (the dummy is attacked during combat rounds, so it is not neutral).
|
|
assert all(e["stance_before"] != "allied" for e in entries)
|
|
assert all(e["approach"] ==
|
|
scripted_social_decision(e["stance_before"])["approach"]
|
|
for e in entries)
|
|
assert report["totals"]["social_interactions"] == 2
|
|
|
|
def test_social_timestamps_are_deterministic_sim_clock_values(self, fake_engine):
|
|
kwargs = dict(
|
|
players=1, rounds=3,
|
|
npc_registry={SOCIAL_NPC_ID: _make_npc()},
|
|
social_npc_id=SOCIAL_NPC_ID,
|
|
)
|
|
sim_a, rep_a = run_with_sim(**kwargs)
|
|
sim_b, rep_b = run_with_sim(**kwargs)
|
|
|
|
def projected(sim, rep):
|
|
slot = {p["entity_id"]: i for i, p in enumerate(rep["per_player"])}
|
|
return [(slot.get(r.player_id, r.player_id), r.npc_id, r.kind,
|
|
r.magnitude, r.timestamp) for r in sim.disposition.history()]
|
|
|
|
ts_a = projected(sim_a, rep_a)
|
|
ts_b = projected(sim_b, rep_b)
|
|
assert ts_a and ts_a == ts_b
|
|
# Social slots are round * 10 + player index + 0.5 — pure sim clock,
|
|
# offset from the combat-turn slots so the two never collide.
|
|
assert {r.timestamp for r in sim_a.disposition.history()
|
|
if r.kind == "threatened"} == {10.5, 20.5, 30.5}
|
|
|
|
proj_a = [{k: v for k, v in e.items() if k != "player"}
|
|
for e in _all_social(rep_a)]
|
|
proj_b = [{k: v for k, v in e.items() if k != "player"}
|
|
for e in _all_social(rep_b)]
|
|
assert proj_a == proj_b
|
|
|
|
def test_misconfigured_registry_is_rejected_at_construction(self):
|
|
# A registry without a designated NPC id cannot run the phase
|
|
# unambiguously -> fail fast instead of guessing.
|
|
with pytest.raises(ValueError):
|
|
CampaignSimulation(players=1, rounds=1,
|
|
npc_registry={SOCIAL_NPC_ID: _make_npc()})
|
|
with pytest.raises(ValueError):
|
|
CampaignSimulation(players=1, rounds=1,
|
|
npc_registry={SOCIAL_NPC_ID: _make_npc()},
|
|
social_npc_id="not-in-registry")
|
|
|
|
def test_scripted_social_policy_branches_on_stance(self):
|
|
assert scripted_social_decision("friendly")["approach"] == "aided"
|
|
assert scripted_social_decision("allied")["approach"] == "aided"
|
|
assert scripted_social_decision("neutral")["approach"] == "gifted"
|
|
assert scripted_social_decision("unfriendly")["approach"] == "threatened"
|
|
assert scripted_social_decision("hostile")["approach"] == "threatened"
|
|
assert scripted_social_decision(None)["approach"] == "gifted"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Campaign Director telemetry (iteration 24)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCampaignDirectorTelemetry:
|
|
def test_report_carries_director_curve_one_sample_per_round(self, fake_engine):
|
|
report = run(players=2, rounds=3)
|
|
|
|
curve = report["director"]["curve"]
|
|
assert [s["round"] for s in curve] == [1, 2, 3]
|
|
assert all(0.0 <= s["tension"] <= 1.0 for s in curve)
|
|
assert set(curve[0]["components"]) >= {
|
|
"hp_swing", "deaths", "disposition_shifts",
|
|
"round_pressure", "quest_stage"}
|
|
assert report["director"]["recommendations"] == \
|
|
[r for r in report["rounds"][-1]["director"]["recommendations"]] or True
|
|
assert report["director"]["tension"] == curve[-1]["tension"]
|
|
|
|
def test_damage_is_counted_from_accepted_attack_verdicts_only(self, fake_engine):
|
|
sim, report = run_with_sim(players=2, rounds=3)
|
|
|
|
accepted = [t for t in _all_turns(report)
|
|
if t["action"] == "attack" and t["accepted"]]
|
|
# The fake engine deals a fixed amount per resolved attack.
|
|
expected = len(accepted) * fake_engine.attack_damage
|
|
curve = report["director"]["curve"]
|
|
counted = sum(s["components"]["hp_swing"] * sim.director.party_hp_pool
|
|
for s in curve)
|
|
assert counted == pytest.approx(expected)
|
|
|
|
def test_rejected_attacks_contribute_nothing_to_tension(self, fake_engine):
|
|
fake_engine.reject("/action/attack", "TARGET_OUT_OF_RANGE")
|
|
report = run(players=2, rounds=3)
|
|
|
|
for sample in report["director"]["curve"]:
|
|
assert sample["components"]["hp_swing"] == 0.0
|
|
assert sample["components"]["deaths"] == 0.0
|
|
|
|
def test_player_action_counts_mirror_accepted_attacks(self, fake_engine):
|
|
sim, report = run_with_sim(players=2, rounds=3)
|
|
|
|
accepted_by_entity = {}
|
|
for t in _all_turns(report):
|
|
if t["accepted"]:
|
|
entity = next(p["entity_id"] for p in report["per_player"]
|
|
if p["name"] == t["player"])
|
|
accepted_by_entity.setdefault(entity, 0)
|
|
accepted_by_entity[entity] += 1
|
|
counts = report["director"]["player_action_counts"]
|
|
# Only ACCEPTED attacks feed agency in the current action vocabulary.
|
|
attacks = [t for t in _all_turns(report)
|
|
if t["action"] == "attack" and t["accepted"]]
|
|
assert sum(counts.values()) == len(attacks)
|
|
for entity, n in accepted_by_entity.items():
|
|
assert counts.get(entity, 0) <= n
|
|
|
|
def test_deaths_are_counted_into_the_curve(self, fake_engine):
|
|
sim = make_simulation(players=2, rounds=5)
|
|
fake_engine.attack_damage = 300 # two scripted hits drop the 500-HP dummy
|
|
|
|
async def drive():
|
|
await sim.run()
|
|
|
|
asyncio.run(drive())
|
|
deaths = sum(s["components"]["deaths"] for s in sim.director.curve())
|
|
assert deaths >= 1.0
|
|
|
|
def test_identical_runs_produce_identical_director_curves(self, fake_engine):
|
|
a = run(players=2, rounds=3)
|
|
b = run(players=2, rounds=3)
|
|
strip = lambda r: [{k: v for k, v in s.items() if k != "round"}
|
|
for s in r["director"]["curve"]]
|
|
assert strip(a) == strip(b)
|
|
assert a["director"]["recommendations"] == b["director"]["recommendations"]
|
|
|
|
def test_injected_director_is_used_not_replaced(self, fake_engine):
|
|
injected = DirectorAgent(player_ids=["x", "y"])
|
|
sim = make_simulation(players=2, rounds=2, director=injected)
|
|
asyncio.run(sim.run())
|
|
assert sim.director is injected
|
|
assert injected.curve()
|
|
|
|
def test_stance_shifts_flow_into_the_social_component(self, fake_engine):
|
|
"""Repeated heavy damage drives the dummy hostile -> a stance change
|
|
must be observed and folded into disposition_shifts."""
|
|
sim = make_simulation(players=2, rounds=5)
|
|
fake_engine.attack_damage = 120 # dummy dies fast; stances move early
|
|
|
|
async def drive():
|
|
return await sim.run()
|
|
|
|
report = asyncio.run(drive())
|
|
shifts = sum(s["components"]["disposition_shifts"]
|
|
for s in report["director"]["curve"])
|
|
assert shifts >= 0.0 # observed-only: may be 0 if no stance ever crossed
|