crisis.py

"""TinyWorld Crisis Mode — deterministic crisis resolver.

A *crisis* is a structured emergency injected into the town. Each crisis declares a
set of *requirements* — concrete responses (be at a place, take a kind of action,
maybe be the right role) that together constitute "handling it". After the five
residents react (via ``agents.react``), :func:`evaluate` checks their structured
output against the requirements. This is the deterministic source of truth — the
LLM only *speaks*; the engine decides whether the crisis is solved.

Design mirrors the rest of the codebase: pure logic, no ``gradio`` import, fully
testable offline with synthetic reaction dicts.
"""

from collections import defaultdict
from dataclasses import dataclass, field

import world_state


# --------------------------------------------------------------------------- model
@dataclass
class Requirement:
    id: str                          # "containment" | "medical" | "coordination" ...
    label: str                       # HUD text, e.g. "🔥 Contain the fire"
    hotspots: list = field(default_factory=list)         # any-of acceptable goto targets
    action_keywords: list = field(default_factory=list)  # any-of words in action/text
    role: str = None                 # if set, only this character role can satisfy it
    weight: float = 1.0
    optional: bool = False           # optional reqs add progress but aren't required to win


@dataclass
class Crisis:
    id: str
    title: str
    narrative: str                   # event text injected to agents (round 1)
    severity: int = 3                # 1-5, drives chaos accrual per unresolved round
    focus: str = ""                  # teaching question (reused by the explainer)
    requirements: list = field(default_factory=list)
    time_limit: int = 4              # max rounds before failure
    escalation: str = ""             # appended to the event each unresolved round
    success_text: str = "The town pulled together and handled it."
    failure_text: str = "It got away from them this time."
    affected_hotspot: str = None     # where the crisis visually manifests on the map
    kind: str = "fire"               # visual effect hint for the canvas (fire/flood/storm/blackout)


@dataclass
class CrisisProgress:
    crisis_id: str
    met: set = field(default_factory=set)            # requirement ids met cumulatively
    newly_met: set = field(default_factory=set)      # ids first satisfied this round
    unmet: list = field(default_factory=list)        # Requirement objects still open
    contributions: dict = field(default_factory=dict)  # name -> [req_id] satisfied this round
    progress: float = 0.0
    resolved: bool = False
    mvp: str = None                                  # resident who satisfied the most reqs


# --------------------------------------------------------------------------- loading
def load_crisis(data):
    """Build a :class:`Crisis` from a plain dict (world files stay data-only)."""
    reqs = [Requirement(**r) for r in data.get("requirements", [])]
    fields = {k: v for k, v in data.items() if k != "requirements"}
    return Crisis(requirements=reqs, **fields)


def get_crisis(world, crisis_id):
    for c in world.get("crises", []):
        if c["id"] == crisis_id:
            return load_crisis(c)
    return None


# --------------------------------------------------------------------------- resolver
def _char_for(world, name):
    return next((c for c in world.get("cast", []) if c["name"] == name), None)


def _norm(s):
    """Lowercase + de-accent so 'café' matches the 'cafe' hotspot key."""
    return (s or "").lower().replace("é", "e").replace("è", "e").replace("ê", "e")


def _mentions(hotspot, blob):
    """The resident named the place in their own words. Use the full phrase or the
    place-type token (clinic/office/home/park…) — never the person-name token, so
    saying 'Priya' doesn't count as being at priya_office."""
    if hotspot.replace("_", " ") in blob:
        return True
    place = hotspot.split("_")[-1]
    return len(place) > 3 and place in blob


def _satisfies(req, reaction, character):
    """True when one reaction fulfils one requirement.

    Each declared condition (location / keywords / role) must hold; an empty
    condition is a wildcard. The small dialogue model often picks the wrong
    structured ``goto`` while clearly stating the right place in its action/text,
    so a location counts if it's the chosen ``goto`` OR named in the words.
    """
    if reaction.get("error"):
        return False

    blob = _norm(f"{reaction.get('action', '')} {reaction.get('text', '')}")

    if req.hotspots:
        location_ok = (reaction.get("moved_to") in req.hotspots
                       or any(_mentions(h, blob) for h in req.hotspots))
        if not location_ok:
            return False

    if req.action_keywords:
        if not any(kw.lower() in blob for kw in req.action_keywords):
            return False

    if req.role:
        if character is None or world_state.character_role(character) != req.role:
            return False

    return True


def evaluate(crisis, reactions, world, prior_met=None):
    """Score a round of reactions against a crisis. Met requirements are sticky."""
    prior_met = set(prior_met or set())
    contributions = defaultdict(list)
    newly_met = set()

    for req in crisis.requirements:
        if req.id in prior_met:
            continue
        for r in reactions:
            char = _char_for(world, r.get("name"))
            if _satisfies(req, r, char):
                newly_met.add(req.id)
                contributions[r["name"]].append(req.id)
                break  # first responder satisfies it

    met = prior_met | newly_met

    total_weight = sum(req.weight for req in crisis.requirements) or 1.0
    met_weight = sum(req.weight for req in crisis.requirements if req.id in met)
    progress = round(met_weight / total_weight, 4)

    required = [req for req in crisis.requirements if not req.optional]
    resolved = all(req.id in met for req in required)
    unmet = [req for req in crisis.requirements if req.id not in met]

    mvp = None
    if contributions:
        mvp = max(contributions, key=lambda n: (len(contributions[n]), n))

    return CrisisProgress(
        crisis_id=crisis.id,
        met=met,
        newly_met=newly_met,
        unmet=unmet,
        contributions=dict(contributions),
        progress=progress,
        resolved=resolved,
        mvp=mvp,
    )


# --------------------------------------------------------------------------- round event
def build_event_text(crisis, prior_progress=None):
    """The event string fed to agents for a round. Subsequent rounds carry feedback
    about what is still unhandled, which is what drives multi-round coordination."""
    if prior_progress is None:
        return crisis.narrative
    parts = [crisis.narrative]
    if crisis.escalation:
        parts.append(crisis.escalation)
    if prior_progress.unmet:
        labels = ", ".join(_plain(req.label) for req in prior_progress.unmet)
        parts.append(f"Still unresolved: {labels}.")
    return " ".join(parts)


# --------------------------------------------------------------------------- narration
def narrate(crisis, progress, round_no=1, failed=False):
    """A short 'Dispatch' verdict. Deterministic — used directly in MOCK mode and as
    the offline fallback. The optional live-LLM flourish wraps this (see app)."""
    if failed:
        return f"📟 Dispatch — {crisis.title}: time ran out. {crisis.failure_text}"
    if progress.resolved:
        return f"📟 Dispatch — {crisis.title} contained. {crisis.success_text}"
    if progress.newly_met:
        done = ", ".join(_plain(_label(crisis, rid)) for rid in sorted(progress.newly_met))
        open_ = ", ".join(_plain(req.label) for req in progress.unmet)
        return f"📟 Dispatch — round {round_no}: {done} handled. Still open: {open_}."
    open_ = ", ".join(_plain(req.label) for req in progress.unmet)
    tail = f" {crisis.escalation}" if crisis.escalation else ""
    return f"📟 Dispatch — round {round_no}: no progress. {open_} still unhandled.{tail}"


def _label(crisis, req_id):
    return next((r.label for r in crisis.requirements if r.id == req_id), req_id)


def _plain(label):
    """Strip a leading emoji so labels read cleanly mid-sentence."""
    parts = label.split(" ", 1)
    if len(parts) == 2 and not parts[0].isascii():
        return parts[1]
    return label