scrypt/warden/director.py

"""The difficulty director: cruel in flavor, bounded in math.

A deterministic monitor watches the fight. When the player is cruising it
offers the Warden a small menu of legal interventions; when the player is
being crushed it can secretly soften. The LLM (when present) only picks
from the menu — every option was pre-validated, and a seeded RNG stands in
when there is no model or the model answers nonsense.
"""

from __future__ import annotations

import random
from dataclasses import dataclass, field

from scrypt.engine.combat import LANES, CombatState
from scrypt.inference.backend import Backend

from .context import build_messages, combat_digest
from .guardrails import PER_FIGHT_LIMITS, ActionBudget
from .harness import Harness, Tool

# Cards the director may add to the queue, weakest first. Indexed by how
# dominant the player is — never more than one tier of pushback at a time.
REINFORCEMENTS = ["reaper", "audit-daemon", "kill-signal"]

DOMINATING, EVEN, CRUSHED = "dominating", "even", "crushed"


def board_power(row) -> int:
    return sum(c.power for c in row if c is not None)


def assess(state: CombatState) -> str:
    """Rolling dominance read on the fight, from the player's side."""
    score = (
        state.scale * 2
        + (board_power(state.player_row) - board_power(state.foe_row))
        + len(state.hand) / 2
    )
    if score >= 6:
        return DOMINATING
    if score <= -6:
        return CRUSHED
    return EVEN


@dataclass
class Intervention:
    action: str
    detail: str
    taunt_moment: str  # handed to the voice for a live line


@dataclass
class Director:
    content: object  # data.Content (for reinforcement card specs)
    rng: random.Random = field(default_factory=random.Random)
    backend: Backend | None = None
    budget: ActionBudget = field(default_factory=lambda: ActionBudget(dict(PER_FIGHT_LIMITS)))
    # Rises with wins on the uptime ledger (capped in legacy.audit_level):
    # a respected adversary gets pushed back against more often.
    audit_level: int = 0

    def new_fight(self) -> None:
        self.budget.reset()
        self.budget.limits["upgrade_script"] = (
            PER_FIGHT_LIMITS["upgrade_script"] + self.audit_level
        )

    # -------------------------------------------------------------- menus

    def _legal_options(self, state: CombatState, mood: str) -> dict[str, dict]:
        options: dict[str, dict] = {}
        if mood is DOMINATING:
            lanes = [
                i for i in range(LANES)
                if state.player_row[i] is not None and state.player_row[i].power > 0
            ]
            if lanes and self.budget.spent.get("tamper_player_deck", 0) < 1:
                strongest = max(lanes, key=lambda i: state.player_row[i].power)
                options["throttle"] = {"lane": strongest}
            empty = [i for i in range(LANES) if state.foe_queue[i] is None]
            if empty:
                tier = min(2, max(0, state.scale // 2))
                options["reinforce"] = {
                    "lane": self.rng.choice(empty),
                    "card": REINFORCEMENTS[tier],
                }
        elif mood is CRUSHED:
            queued = [
                i for i in range(LANES)
                if state.foe_queue[i] is not None and state.foe_queue[i].power >= 2
            ]
            if queued:
                worst = max(queued, key=lambda i: state.foe_queue[i].power)
                options["withdraw"] = {"lane": worst}
        return options

    # ------------------------------------------------------------ choosing

    async def consider(self, state: CombatState) -> Intervention | None:
        """Call after the bell, before the player's next turn."""
        if state.turn < 2:
            return None  # let the opening breathe
        mood = assess(state)
        if mood is EVEN:
            return None
        options = self._legal_options(state, mood)
        if not options:
            return None
        budget_key = "secret_mercy" if mood is CRUSHED else "upgrade_script"
        if not self.budget.try_spend(budget_key):
            return None
        choice = await self._choose(state, mood, list(options))
        return self._apply(state, choice, options[choice])

    async def _choose(self, state: CombatState, mood: str, names: list[str]) -> str:
        if self.backend is None or len(names) == 1:
            return self.rng.choice(names)
        picked: list[str] = []
        tool = Tool(
            name="intervene",
            description="choose one intervention",
            schema={"properties": {"action": {"enum": names}}, "required": ["action"]},
            handler=lambda args: picked.append(args["action"]) or "chosen",
        )
        frame = (
            f"The player is {mood}. Choose one intervention from {names} "
            'by calling the tool. Pick what stings most.'
        )
        harness = Harness(self.backend, [tool], max_steps=2, max_tokens=120)
        await harness.run(build_messages(frame, digest=combat_digest(state)))
        return picked[0] if picked else self.rng.choice(names)

    # ------------------------------------------------------------ applying

    def _apply(self, state: CombatState, action: str, args: dict) -> Intervention:
        if action == "throttle":
            if self.budget.try_spend("tamper_player_deck"):
                detail = state.throttle_player_card(args["lane"])
                card = state.player_row[args["lane"]].spec.id
                return Intervention(
                    "throttle", detail,
                    f"you just quietly weakened the player's {card}; gloat without "
                    "explaining exactly what you did",
                )
        if action == "reinforce":
            spec = self.content.card(args["card"])
            detail = state.reinforce_queue(args["lane"], spec)
            return Intervention(
                "reinforce", detail,
                f"you just scheduled an extra {spec.id} against the player",
            )
        if action == "withdraw":
            detail = state.withdraw_queue(args["lane"])
            return Intervention(
                "withdraw", detail,
                "you quietly removed a threat because the player is being crushed; "
                "say something that sounds like boredom, not kindness",
            )
        # Budget refused late: report nothing happened.
        return Intervention("none", "no-op", "")