"""Autopass execution + rule management (plan §4.4) — automation as a teammate, scoped to its smallest useful form. A rule belongs to ONE player in ONE match: "when a ball is passed to me from , auto-pass it to ". Execution is exactly ONE hop: rules are evaluated ONLY against the primary HUMAN touch (serve / pass / spike), and the auto touch a rule records (kind "auto", via_rule_id set) is never fed back into matching — loop-proof by construction, not by a depth counter. Honesty is the rally timeline's job: the auto touch keeps its via_rule_id so every payload renders it with the bolt glyph. Trigger semantics (matched against the primary touch's from_player): - from_subteam the passer is on a team the rule's owner LEADS — their node, or any ancestor of it, has the owner as boundary player (topology.led_by). Inclusive of the owner's own node: Zeina IS Adi's sub-team. - from_any any passer. - from_player exactly rule.trigger_player_id. An authorless movement (from_player NULL) never triggers anything — automation reacts to a person's pass, not to system bookkeeping. Called INSIDE the owning service's transaction (services.touches / services.board), same convention as services.notifications: never commits, the caller's commit covers the auto touch + assignment change. """ from __future__ import annotations from sqlalchemy import select from sqlalchemy.orm import Session from ..models import AutopassRule, Ball, Court, Match, Player, Touch from . import projection as projection_service from .board import BoardError from .perspective import MatchTopology, PerspectiveLens TRIGGERS = ("from_subteam", "from_any", "from_player") # ------------------------------------------------------- rule management def list_rules(db: Session, match_id, owner: Player) -> list[AutopassRule]: """The owner's own rule(s) in this match (0 or 1 — schema unique). Owner-scoped by construction: the query is keyed on the caller's identity, so there is no path to another player's rules.""" return ( db.execute( select(AutopassRule).where( AutopassRule.match_id == match_id, AutopassRule.owner_player_id == owner.id, ) ) .scalars() .all() ) def upsert_rule(db: Session, match: Match, owner: Player, data: dict) -> AutopassRule: """Create or replace the owner's ONE rule for this match (§4.4: two dropdowns and a toggle — PUT is the natural verb). Validation: - trigger must be a known TRIGGERS value; trigger_player_id is required for from_player and cleared otherwise (a stored stray trigger_player on from_any would be dead config lying in wait); - the target — and a from_player trigger's player — must be VISIBLE to the owner (§4.3b: automation is just a standing pass, and you can only pass to players on your courts); - the target may not be the owner (a self-target could only ever re-assign the ball in place — refuse it loudly instead of storing a rule that does nothing). Commits (this IS the triggering write, mirroring mark_read's convention in services.notifications). """ trigger = data.get("trigger") if trigger not in TRIGGERS: raise BoardError( 422, "INVALID_TRIGGER", f"{trigger!r} is not a valid autopass trigger", f"Valid values: {', '.join(TRIGGERS)}", ) trigger_player_id = data.get("trigger_player_id") if trigger == "from_player" and trigger_player_id is None: raise BoardError( 422, "TRIGGER_PLAYER_REQUIRED", "A from_player rule needs the player it watches for.", "Send trigger_player_id, or use trigger from_any / from_subteam.", ) if trigger != "from_player": trigger_player_id = None target_player_id = data["target_player_id"] if target_player_id == owner.id: raise BoardError( 422, "INVALID_AUTOPASS_TARGET", "An autopass cannot target yourself.", "The rule would re-assign the ball in place; pick a teammate.", ) lens = PerspectiveLens(MatchTopology(db, match.id), owner.id) check_ids = [target_player_id] + ( [trigger_player_id] if trigger_player_id is not None else [] ) projection_service.ensure_targets_visible(lens, check_ids, BoardError) rule = db.execute( select(AutopassRule).where( AutopassRule.match_id == match.id, AutopassRule.owner_player_id == owner.id, ) ).scalar_one_or_none() if rule is None: rule = AutopassRule(match_id=match.id, owner_player_id=owner.id) db.add(rule) rule.trigger = trigger rule.trigger_player_id = trigger_player_id rule.target_player_id = target_player_id rule.active = bool(data.get("active", True)) db.commit() db.refresh(rule) return rule # ---------------------------------------------------------- execution def rules_by_owner(db: Session, match_id) -> dict: """Active rules for a match, keyed by owner_player_id (at most one per owner — schema unique constraint).""" rows = ( db.execute( select(AutopassRule).where( AutopassRule.match_id == match_id, AutopassRule.active.is_(True) ) ) .scalars() .all() ) return {r.owner_player_id: r for r in rows} def rule_matches(rule: AutopassRule, topology: MatchTopology, from_player) -> bool: if from_player is None: return False if rule.trigger == "from_any": return True if rule.trigger == "from_player": return rule.trigger_player_id == from_player if rule.trigger == "from_subteam": return topology.led_by(rule.owner_player_id, from_player) return False # unknown trigger value = inert, never an error at play time def apply_autopass( db: Session, ball: Ball, court: Court, primary_touch: Touch, topology: MatchTopology, ) -> list[Touch]: """Fire matching rules for the players the PRIMARY touch just handed the ball to. For each receiver with a matching active rule: swap the receiver for the rule's target in action_to (dedupe, order preserved) and record ONE auto touch from the rule's owner. Returns the auto touches (usually zero or one) for the caller to publish. Never recurses: the targets an auto touch assigns are not re-checked (§4.4's one-hop guarantee — the test where Rian also has a rule that must NOT fire on the auto hop is exactly this line's contract). """ from . import touches as touch_service # local import avoids cycle if primary_touch.via_rule_id is not None or primary_touch.kind == "auto": return [] # defense in depth; callers only pass human touches receivers = list(primary_touch.to_players or []) if not receivers: return [] rules = rules_by_owner(db, court.match_id) auto_touches: list[Touch] = [] for receiver in receivers: rule = rules.get(receiver) if rule is None: continue if rule.target_player_id == receiver: continue # self-target is a stored no-op; never bounce in place if not rule_matches(rule, topology, primary_touch.from_player): continue owner = db.get(Player, receiver) if owner is None: continue # Swap receiver -> target in the CURRENT assignment (a prior fired # rule in this same loop may already have reshaped it). new_action_to = [] for pid in ball.action_to or []: swapped = rule.target_player_id if pid == receiver else pid if swapped not in new_action_to: new_action_to.append(swapped) ball.action_to = new_action_to auto_touches.append( touch_service.record_touch( db, ball, "auto", owner, [rule.target_player_id], via_rule_id=rule.id, court=court, ) ) return auto_touches