diff --git a/packages/domain/src/sagasmith_dnd/combat_engine.py b/packages/domain/src/sagasmith_dnd/combat_engine.py index c0296d63..44000e6d 100644 --- a/packages/domain/src/sagasmith_dnd/combat_engine.py +++ b/packages/domain/src/sagasmith_dnd/combat_engine.py @@ -2118,7 +2118,7 @@ def preflight_attack( or candidate_position is None or _grid_distance(attacker_position, candidate_position) > 5 or not _are_hostile(candidate, attacker) - or not _can_see(candidate, attacker) + or not can_see(candidate, attacker) or _condition_set(candidate.get("conditions")) & INCAPACITATING_STATE_IDS ): continue @@ -2152,10 +2152,10 @@ def preflight_attack( target.get("conditions") or actor_sheet(target).get("conditions") ) attacker_can_see_target = bool( - context.get("attacker_can_see_target", _can_see(attacker, target)) + context.get("attacker_can_see_target", can_see(attacker, target)) ) target_can_see_attacker = bool( - context.get("target_can_see_attacker", _can_see(target, attacker)) + context.get("target_can_see_attacker", can_see(target, attacker)) ) if not target_can_see_attacker: context["advantage"] = True @@ -2205,7 +2205,7 @@ def preflight_attack( combatant for combatant in encounter.get("combatants", []) if str(combatant.get("actor_id") or "") in fear_sources - and _can_see(attacker, combatant) + and can_see(attacker, combatant) ] if visible_sources: context["disadvantage"] = True @@ -4132,7 +4132,7 @@ def spend_movement( raise CombatEngineError("Aggressive target must be a living combatant") if not _are_hostile(combatant, aggressive_target): raise CombatEngineError("Aggressive target is no longer hostile") - if not _can_see(combatant, aggressive_target): + if not can_see(combatant, aggressive_target): raise CombatEngineError("Aggressive target is no longer visible") conditions = _condition_set(combatant.get("conditions")) if willing_movement and conditions & { @@ -4343,7 +4343,7 @@ def spend_movement( "frightened movement source is not in the encounter", missing=("frightened_source_combatant",), ) - if not _can_see(combatant, fear_source): + if not can_see(combatant, fear_source): continue fear_source_position = _position(fear_source.get("position")) if fear_source_position is None: @@ -5515,7 +5515,7 @@ def settle_core_activity_effect( raise CombatEngineError("Aggressive target must be a living combatant") if not _are_hostile(combatant, target): raise CombatEngineError("Aggressive target must be hostile") - if not _can_see(combatant, target): + if not can_see(combatant, target): raise CombatEngineError("Aggressive target must be visible to the Orc") flags = dict(combatant.get("turn_flags") or {}) if "aggressive_movement" in flags: @@ -7052,7 +7052,7 @@ def _disengaged(combatant: dict[str, Any]) -> bool: def _can_make_opportunity_attack(threat: dict[str, Any], moving: dict[str, Any]) -> bool: if threat.get("actor_id") == moving.get("actor_id"): return False - if not _can_see(threat, moving): + if not can_see(threat, moving): # Whether a particular creature can perceive a hidden or invisible # mover is a DM fact unless visible_to_actor_ids records it explicitly. return False @@ -7084,7 +7084,7 @@ def _are_hostile(left: dict[str, Any], right: dict[str, Any]) -> bool: return {left_disposition, right_disposition} == {"hostile", "friendly"} -def _can_see(viewer: dict[str, Any], subject: dict[str, Any]) -> bool: +def can_see(viewer: dict[str, Any], subject: dict[str, Any]) -> bool: """Resolve only recorded visibility, defaulting ordinary creatures to visible.""" viewer_conditions = ( viewer.get("conditions") diff --git a/packages/domain/tests/test_combat_engine.py b/packages/domain/tests/test_combat_engine.py index 113c4909..feb2fec8 100644 --- a/packages/domain/tests/test_combat_engine.py +++ b/packages/domain/tests/test_combat_engine.py @@ -26,6 +26,7 @@ available_actions, available_attack_defenses, available_reactions, + can_see, consume_weapon_mastery_attack_effects, current_combatant, damage_amount_after_reduction, @@ -5354,6 +5355,28 @@ def test_hidden_mover_does_not_automatically_reveal_itself_with_a_reaction_windo assert available_reactions(moved, "threat") == [] +def test_recorded_visibility_is_authoritative_for_sight_checks() -> None: + viewer = _actor("viewer") + subject = _actor("subject") + + assert can_see(viewer, subject) is True + + excluded = deepcopy(subject) + excluded["visible_to_actor_ids"] = [] + assert can_see(viewer, excluded) is False + + concealed = deepcopy(subject) + concealed["hidden"] = True + concealed["conditions"] = ["invisible"] + assert can_see(viewer, concealed) is False + concealed["visible_to_actor_ids"] = ["viewer"] + assert can_see(viewer, concealed) is True + + blinded = deepcopy(viewer) + blinded["sheet"]["conditions"] = ["blinded"] + assert can_see(blinded, concealed) is False + + def test_recorded_visibility_can_open_reaction_window_for_invisible_mover() -> None: mover = _actor("mover") mover["sheet"]["conditions"] = ["invisible"] diff --git a/packages/mcp/src/sagasmith_dnd_mcp/server.py b/packages/mcp/src/sagasmith_dnd_mcp/server.py index c13fbbf3..570204e5 100644 --- a/packages/mcp/src/sagasmith_dnd_mcp/server.py +++ b/packages/mcp/src/sagasmith_dnd_mcp/server.py @@ -192,6 +192,7 @@ available_actions, available_attack_defenses, available_reactions, + can_see, consume_weapon_mastery_attack_effects, current_combatant, damage_amount_after_reduction, @@ -11044,8 +11045,8 @@ def validate_spell_creature_target( raise CombatEngineError("spell caster is not in this encounter") if target is None: raise CombatEngineError(f"spell target is not in this encounter: {target_id}") - conditions = {str(item).casefold() for item in target.get("conditions", [])} - if "dead" in conditions: + target_conditions = {str(item).casefold() for item in target.get("conditions", [])} + if "dead" in target_conditions: raise CombatEngineError("a dead combatant is not a creature target") distance = combat_distance(caster.get("position"), target.get("position")) if distance is None: @@ -11063,9 +11064,7 @@ def validate_spell_creature_target( if range_kind != "self" and distance > maximum: raise CombatEngineError("spell target is outside range") targeting = dict(resolution.get("targeting") or {}) - concealed = bool(target.get("hidden", False)) or "invisible" in conditions - visible_to = {str(item) for item in target.get("visible_to_actor_ids") or []} - if targeting.get("requires_sight") and concealed and caster_id not in visible_to: + if targeting.get("requires_sight") and not can_see(caster, target): raise CombatEngineError("spell requires a target the caster can see") creature_type = str( characters.get(target_id).sheet.get("progression", {}).get("species") or "" diff --git a/packages/mcp/tests/test_structured_spell_mcp.py b/packages/mcp/tests/test_structured_spell_mcp.py index 274d4469..347c731d 100644 --- a/packages/mcp/tests/test_structured_spell_mcp.py +++ b/packages/mcp/tests/test_structured_spell_mcp.py @@ -308,6 +308,31 @@ async def _campaign_with_combat( return campaign["id"], started["campaign_revision"], actors +async def _campaign_actor_snapshot(server, campaign_id: str, actor_ids: list[str]) -> dict: + campaign = await _call( + server, + "campaign_query", + { + "view": "get", + "payload": {"campaign_id": campaign_id}, + "principal_id": "system:local", + }, + ) + actors = [ + await _call( + server, + "character_query", + { + "view": "get", + "payload": {"character_id": actor_id}, + "principal_id": "system:local", + }, + ) + for actor_id in actor_ids + ] + return {"campaign": campaign, "actors": actors} + + def _deterministic_rolls(monkeypatch) -> None: monkeypatch.setattr( server_module, @@ -399,6 +424,221 @@ async def exercise() -> None: asyncio.run(exercise()) +def test_sight_required_spell_rejects_blinded_caster_without_writes( + tmp_path: Path, + monkeypatch, +) -> None: + roll_expressions: list[str] = [] + + def tracked_roll(expression: str): + roll_expressions.append(expression) + return engine_roll(expression, rng=random.Random(7)) + + monkeypatch.setattr(server_module, "roll", tracked_roll) + + async def exercise() -> None: + server = create_server(_config(tmp_path)) + caster = default_character_sheet() + caster["conditions"] = ["blinded"] + caster["spellcasting"].update(ability="wisdom", spell_slots=_slot(1, 2)) + healing_word = _spell("Healing Word", 1, casting_time="1 bonus action", range_ft=60) + cure_wounds = _spell("Cure Wounds", 1, casting_time="1 action", range_ft=5) + caster["content"]["spells"] = [healing_word, cure_wounds] + caster["effects"] = [ + { + "id": "existing-concentration", + "name": "Existing concentration", + "kind": "concentration", + "source": "spell.cast", + "source_spell_id": "test.spell.existing-concentration", + "active": True, + "concentration": True, + "duration": {"period": "minute", "remaining": 10}, + "changes": [], + "description": "", + } + ] + target = default_character_sheet() + target["combat"]["hp"] = {"value": 1, "max": 20, "temp": 0} + campaign_id, revision, actors = await _campaign_with_combat( + server, + [("Blinded cleric", caster), ("Ally", target)], + positions=[(0, 0), (1, 0)], + ) + actor_ids = [item["id"] for item in actors] + before = await _campaign_actor_snapshot( + server, + campaign_id, + actor_ids, + ) + rejected_arguments = { + "campaign_id": campaign_id, + "actor_id": actors[0]["id"], + "spell_id": healing_word["id"], + "cast_level": 1, + "declaration": {"target_id": actors[1]["id"]}, + "expected_revision": revision, + "idempotency_key": "blinded-healing-word", + } + + for _attempt in range(2): + with pytest.raises(Exception, match="spell requires a target the caster can see"): + await _raw(server, "combat_cast_spell", rejected_arguments) + + after = await _campaign_actor_snapshot( + server, + campaign_id, + actor_ids, + ) + assert after == before + assert roll_expressions == [] + + non_sight_cast = await _raw( + server, + "combat_cast_spell", + { + **rejected_arguments, + "spell_id": cure_wounds["id"], + "idempotency_key": "blinded-cure-wounds", + }, + ) + assert non_sight_cast["status"] == "committed" + assert non_sight_cast["campaign_revision"] == revision + 1 + assert roll_expressions == ["1d8"] + caster_after_cast = await _call( + server, + "character_query", + { + "view": "get", + "payload": {"character_id": actors[0]["id"]}, + "principal_id": "system:local", + }, + ) + concentration = next( + item + for item in caster_after_cast["sheet"]["effects"] + if item["id"] == "existing-concentration" + ) + assert concentration["active"] is True + + asyncio.run(exercise()) + + +def test_sight_required_spell_honors_authoritative_visibility_acl( + tmp_path: Path, + monkeypatch, +) -> None: + roll_expressions: list[str] = [] + + def tracked_roll(expression: str): + roll_expressions.append(expression) + return engine_roll(expression, rng=random.Random(7)) + + monkeypatch.setattr(server_module, "roll", tracked_roll) + + async def exercise() -> None: + server = create_server(_config(tmp_path)) + caster = default_character_sheet() + caster["spellcasting"].update(ability="wisdom", spell_slots=_slot(1)) + healing_word = _spell("Healing Word", 1, casting_time="1 bonus action", range_ft=60) + caster["content"]["spells"] = [healing_word] + target = default_character_sheet() + target["combat"]["hp"] = {"value": 1, "max": 20, "temp": 0} + campaign_id, revision, actors = await _campaign_with_combat( + server, + [("Cleric", caster), ("Hidden ally", target)], + positions=[(0, 0), (4, 0)], + ) + excluded = await _raw( + server, + "combat_map_patch", + { + "campaign_id": campaign_id, + "patches": [ + { + "key": "combatant_visibility", + "value": { + "actor_id": actors[1]["id"], + "visible_to_actor_ids": [], + "reason": "The target is fully obscured from the caster.", + }, + } + ], + "expected_revision": revision, + "idempotency_key": "exclude-target", + }, + ) + actor_ids = [item["id"] for item in actors] + before = await _campaign_actor_snapshot( + server, + campaign_id, + actor_ids, + ) + arguments = { + "campaign_id": campaign_id, + "actor_id": actors[0]["id"], + "spell_id": healing_word["id"], + "cast_level": 1, + "declaration": {"target_id": actors[1]["id"]}, + "expected_revision": excluded["campaign_revision"], + "idempotency_key": "acl-healing-word", + } + + with pytest.raises(Exception, match="spell requires a target the caster can see"): + await _raw(server, "combat_cast_spell", arguments) + + after = await _campaign_actor_snapshot( + server, + campaign_id, + actor_ids, + ) + assert after == before + assert roll_expressions == [] + + recorded_visible = await _raw( + server, + "combat_map_patch", + { + "campaign_id": campaign_id, + "patches": [ + { + "key": "combatant_visibility", + "value": { + "actor_id": actors[1]["id"], + "hidden": True, + "visible_to_actor_ids": [actors[0]["id"]], + "reason": "The caster pinpointed the hidden target.", + }, + } + ], + "expected_revision": excluded["campaign_revision"], + "idempotency_key": "record-target-visible", + }, + ) + succeeded = await _raw( + server, + "combat_cast_spell", + { + **arguments, + "expected_revision": recorded_visible["campaign_revision"], + }, + ) + assert succeeded["status"] == "committed" + assert roll_expressions == ["1d4"] + target_after_cast = await _call( + server, + "character_query", + { + "view": "get", + "payload": {"character_id": actors[1]["id"]}, + "principal_id": "system:local", + }, + ) + assert target_after_cast["sheet"]["combat"]["hp"]["value"] > 1 + + asyncio.run(exercise()) + + def test_scorching_ray_cast_locks_then_settles_each_source_bound_attack( tmp_path: Path, monkeypatch ) -> None: