Spaces:
Sleeping
Sleeping
| """TinyWorld Crisis Mode — campaign state machine. | |
| A *campaign* chains crises into a story ("Season 1 · The Long Summer"). Each | |
| *chapter* wraps one crisis with narrative beats and a meta-mystery clue. The town | |
| survives the campaign by keeping its Town Resilience above zero; the finale reveals | |
| that the disasters were never random — the saboteur is the player. | |
| This module is pure orchestration: it never calls the LLM itself. The UI computes | |
| reactions with ``agents.react`` and hands them to :func:`resolve_round`, which keeps | |
| the engine testable offline with scripted reactions. | |
| """ | |
| from dataclasses import dataclass, field | |
| import crisis as crisis_mod | |
| import world_state | |
| # Resilience cost of finishing a crisis at each grade (a clean run barely dents it). | |
| GRADE_COST = {"A": 0, "B": 3, "C": 7, "D": 12, "F": 25} | |
| class Chapter: | |
| id: str | |
| crisis_id: str | |
| intro: str = "" | |
| outro_win: str = "" | |
| outro_loss: str = "" | |
| clue: str = "" | |
| class Campaign: | |
| id: str | |
| title: str | |
| chapters: list = field(default_factory=list) | |
| finale_reveal: str = "" | |
| # --------------------------------------------------------------------------- loading | |
| def get_campaign_def(world): | |
| data = world.get("campaign") | |
| if not data: | |
| return None | |
| chapters = [Chapter(**ch) for ch in data.get("chapters", [])] | |
| return Campaign( | |
| id=data["id"], | |
| title=data["title"], | |
| chapters=chapters, | |
| finale_reveal=data.get("finale_reveal", ""), | |
| ) | |
| def current_chapter(world): | |
| camp = get_campaign_def(world) | |
| if not camp: | |
| return None | |
| idx = world_state.get_campaign(world["id"])["chapter_index"] | |
| if 0 <= idx < len(camp.chapters): | |
| return camp.chapters[idx] | |
| return None | |
| def current_crisis(world): | |
| state = world_state.get_campaign(world["id"]) | |
| if not state.get("crisis_id"): | |
| return None | |
| return crisis_mod.get_crisis(world, state["crisis_id"]) | |
| # --------------------------------------------------------------------------- lifecycle | |
| def start(world): | |
| """Begin the campaign at chapter 1 and inject its crisis.""" | |
| wid = world["id"] | |
| world_state.reset_campaign(wid) | |
| camp = get_campaign_def(world) | |
| if not camp or not camp.chapters: | |
| return view(world) | |
| first = camp.chapters[0] | |
| world_state.campaign_update( | |
| wid, | |
| active=True, | |
| campaign_id=camp.id, | |
| chapter_index=0, | |
| status="playing", | |
| ) | |
| _inject(world, first) | |
| return view(world) | |
| def _inject(world, chapter): | |
| wid = world["id"] | |
| world_state.campaign_update( | |
| wid, | |
| crisis_id=chapter.crisis_id, | |
| crisis_round=0, | |
| crisis_met=set(), | |
| last_dispatch="", | |
| ) | |
| world_state.reset_chaos(wid) | |
| def round_event_text(world): | |
| """The event string to feed agents this round (carries prior-round feedback).""" | |
| cr = current_crisis(world) | |
| if not cr: | |
| return "" | |
| state = world_state.get_campaign(world["id"]) | |
| if state["crisis_round"] == 0: | |
| return crisis_mod.build_event_text(cr) | |
| # rebuild a progress-shaped object from the sticky met set for the feedback text | |
| prior = crisis_mod.CrisisProgress( | |
| crisis_id=cr.id, | |
| met=state["crisis_met"], | |
| unmet=[r for r in cr.requirements if r.id not in state["crisis_met"]], | |
| ) | |
| return crisis_mod.build_event_text(cr, prior_progress=prior) | |
| # --------------------------------------------------------------------------- a round | |
| def resolve_round(world, reactions): | |
| """Advance the campaign by one player round given the residents' reactions.""" | |
| wid = world["id"] | |
| state = world_state.get_campaign(wid) | |
| cr = current_crisis(world) | |
| if not cr or state["status"] not in ("playing",): | |
| return {"outcome": "idle", **view(world)} | |
| round_no = state["crisis_round"] + 1 | |
| prog = crisis_mod.evaluate(cr, reactions, world, prior_met=state["crisis_met"]) | |
| world_state.campaign_update(wid, crisis_met=prog.met, crisis_round=round_no) | |
| if prog.mvp: | |
| tally = state["mvp_tally"] | |
| tally[prog.mvp] = tally.get(prog.mvp, 0) + 1 | |
| world_state.campaign_update(wid, mvp_tally=tally) | |
| if prog.resolved: | |
| grade = _grade(round_no, world_state.get_chaos(wid), won=True) | |
| dispatch = crisis_mod.narrate(cr, prog, round_no=round_no) | |
| return _finish_chapter(world, cr, grade, prog, dispatch, won=True) | |
| # not resolved — chaos ramps with each round so the time limit is the real | |
| # budget (campaign owns the chaos gauge; we overwrite any per-event drift). | |
| pressure = min(1.0, (round_no / cr.time_limit) * (0.8 + 0.05 * cr.severity)) | |
| world_state.reset_chaos(wid) | |
| world_state.add_chaos(wid, pressure) | |
| if round_no >= cr.time_limit: | |
| grade = "F" | |
| dispatch = crisis_mod.narrate(cr, prog, round_no=round_no, failed=True) | |
| return _finish_chapter(world, cr, grade, prog, dispatch, won=False) | |
| dispatch = crisis_mod.narrate(cr, prog, round_no=round_no) | |
| world_state.campaign_update(wid, last_dispatch=dispatch) | |
| return {"outcome": "ongoing", "dispatch": dispatch, "progress": prog, "grade": None, **view(world)} | |
| def _finish_chapter(world, cr, grade, prog, dispatch, won): | |
| wid = world["id"] | |
| state = world_state.get_campaign(wid) | |
| camp = get_campaign_def(world) | |
| chapter = current_chapter(world) | |
| resilience = max(0, state["town_resilience"] - GRADE_COST.get(grade, 12)) | |
| grades = state["chapter_grades"] + [{ | |
| "chapter": chapter.id if chapter else cr.id, | |
| "title": cr.title, | |
| "grade": grade, | |
| "rounds": state["crisis_round"], | |
| "mvp": prog.mvp, | |
| "won": won, | |
| }] | |
| clues = state["clues"] + ([chapter.clue] if chapter and chapter.clue else []) | |
| world_state.campaign_update( | |
| wid, | |
| town_resilience=resilience, | |
| chapter_grades=grades, | |
| clues=clues, | |
| last_dispatch=dispatch, | |
| ) | |
| next_index = state["chapter_index"] + 1 | |
| outcome = "won" if won else "lost_crisis" | |
| outro = (chapter.outro_win if won else chapter.outro_loss) if chapter else "" | |
| if resilience <= 0: | |
| world_state.campaign_update(wid, status="lost", crisis_id=None) | |
| return {"outcome": "campaign_lost", "dispatch": dispatch, "outro": outro, | |
| "grade": grade, "progress": prog, **view(world)} | |
| if next_index >= len(camp.chapters): | |
| # campaign cleared — the meta-mystery resolves | |
| world_state.campaign_update(wid, status="finale", chapter_index=next_index, crisis_id=None) | |
| return {"outcome": "finale", "dispatch": dispatch, "outro": outro, "grade": grade, | |
| "reveal": camp.finale_reveal, "progress": prog, **view(world)} | |
| world_state.campaign_update(wid, chapter_index=next_index) | |
| _inject(world, camp.chapters[next_index]) | |
| return {"outcome": outcome, "dispatch": dispatch, "outro": outro, "grade": grade, | |
| "progress": prog, **view(world)} | |
| def _grade(round_no, chaos, won): | |
| if not won: | |
| return "F" | |
| if round_no == 1: | |
| base = "A" | |
| elif round_no == 2: | |
| base = "B" | |
| elif round_no == 3: | |
| base = "C" | |
| else: | |
| base = "D" | |
| # a messy scene (high chaos) knocks the grade down one notch | |
| if chaos >= 0.6 and base in ("A", "B", "C"): | |
| base = {"A": "B", "B": "C", "C": "D"}[base] | |
| return base | |
| # --------------------------------------------------------------------------- snapshot | |
| def view(world): | |
| """A serializable snapshot of campaign + active crisis for the UI/HUD.""" | |
| wid = world["id"] | |
| camp = get_campaign_def(world) | |
| state = world_state.get_campaign(wid) | |
| cr = current_crisis(world) | |
| chapter = current_chapter(world) | |
| crisis_view = None | |
| if cr: | |
| crisis_view = { | |
| "id": cr.id, | |
| "title": cr.title, | |
| "kind": cr.kind, | |
| "affected_hotspot": cr.affected_hotspot, | |
| "severity": cr.severity, | |
| "time_limit": cr.time_limit, | |
| "round": state["crisis_round"], | |
| "focus": cr.focus, | |
| "requirements": [ | |
| {"id": r.id, "label": r.label, "met": r.id in state["crisis_met"], | |
| "optional": r.optional} | |
| for r in cr.requirements | |
| ], | |
| } | |
| return { | |
| "active": state["active"], | |
| "status": state["status"], | |
| "title": camp.title if camp else "", | |
| "chapter_index": state["chapter_index"], | |
| "chapter_count": len(camp.chapters) if camp else 0, | |
| "chapter_intro": chapter.intro if chapter else "", | |
| "town_resilience": state["town_resilience"], | |
| "chaos": round(world_state.get_chaos(wid), 3), | |
| "clues": state["clues"], | |
| "chapter_grades": state["chapter_grades"], | |
| "mvp_tally": state["mvp_tally"], | |
| "last_dispatch": state["last_dispatch"], | |
| "crisis": crisis_view, | |
| } | |