Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions packages/domain/src/sagasmith_dnd/combat_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 & {
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand Down
23 changes: 23 additions & 0 deletions packages/domain/tests/test_combat_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
available_actions,
available_attack_defenses,
available_reactions,
can_see,
consume_weapon_mastery_attack_effects,
current_combatant,
damage_amount_after_reduction,
Expand Down Expand Up @@ -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"]
Expand Down
9 changes: 4 additions & 5 deletions packages/mcp/src/sagasmith_dnd_mcp/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -192,6 +192,7 @@
available_actions,
available_attack_defenses,
available_reactions,
can_see,
consume_weapon_mastery_attack_effects,
current_combatant,
damage_amount_after_reduction,
Expand Down Expand Up @@ -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:
Expand All @@ -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 ""
Expand Down
240 changes: 240 additions & 0 deletions packages/mcp/tests/test_structured_spell_mcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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:
Expand Down