"""Post-LLM validation — the checklist in ARCHITECTURE.md. Any failure means the caller substitutes a fallback. Clamp-able numeric drift is clamped instead of rejected (the player should never lose a live response to a rounding issue). Every rejection is traced with its reason (logs/trace.log) for prompt tuning. """ from __future__ import annotations import re from .schemas import (ANIMATION_TRIGGERS, APPROVED_CLIENTS, BOARD_TONES, EVENT_CATEGORIES, NPC_IDS, ROUND_DIFFICULTIES, SPECIAL_EVENT_TYPES) from .state import GameState from .trace import trace BANNED_PATTERNS = [ r"\b(cancer|tumou?r|terminal|chemotherapy|stroke|heart attack|overdose|" r"suicide|self.?harm|seizure|diagnos)\w*", r"\b(lawsuit|subpoena|plaintiff|defendant|litigation|felony|indictment|" r"class.?action|sue|sued|suing|legal action)\b", r"\b(google|microsoft|apple|amazon|meta|openai|anthropic|tesla|netflix|" r"salesforce|oracle|ibm)\b", r"\b(democrat|republican|election|congress|senate|president(ial)?)\b", ] _BANNED = [re.compile(p, re.IGNORECASE) for p in BANNED_PATTERNS] # everyday workplace tools are fair game for comedy — only the bare company # name (used as a fictional client) is banned. Strip tool collocations before # the company check so "Google Docs" passes but "Google signed us" does not. ALLOWED_TOOLS = re.compile( r"\bgoogle\s+(docs?|sheets?|slides?|drive|calendar|meet|forms?|maps|workspace|chat)\b" r"|\bmicrosoft\s+(word|excel|teams|outlook|powerpoint|office|365|sharepoint)\b" r"|\bapple\s+(watch|store|pay|tv|music|notes)\b" r"|\bamazon\s+(package|delivery|order|prime|parcel|box|web services|aws)\b", re.IGNORECASE) # note: no bare "date" — office prose is full of launch dates and deadlines ROMANCE_WORDS = re.compile( r"\b(dating|romance|romantic|kiss(es|ed|ing)?|crush on|love you|" r"in love|go(ing)? out with|on a date)\b", re.IGNORECASE) # game mechanics must never leak into player-visible prose MECHANICS_LEAK = re.compile( r"\b(morale|relationship_delta|revenue_delta|pocket_money_delta|" r"log_entry|npc_id|special_next_event|setup_animation|cumulative_score|" r"board_tone|morale_(delta|preview)|game state|event_log)\b", re.IGNORECASE) # the most recent rejection reason — read by llm.call_validated to build a # corrective retry nudge so a single model slip doesn't force a fallback LAST_REJECT = None def _reject(kind: str, reason: str) -> None: global LAST_REJECT LAST_REJECT = reason trace("vald", f"REJECT {kind}: {reason}") return None # decorative comic fields are soft-validated separately (_clean_comic_fields), # so they're excluded from the hard banned/leak scans that reject the payload _COMIC_FIELDS = ("image_prompt", "comic_caption") def _text_fields(payload: dict) -> str: return " ".join(str(v) for k, v in payload.items() if isinstance(v, str) and k not in _COMIC_FIELDS) def banned_match(payload: dict) -> str | None: text = ALLOWED_TOOLS.sub(" ", _text_fields(payload)) for pattern in _BANNED: m = pattern.search(text) if m: return m.group(0) return None def _clamp(value, lo, hi): return max(lo, min(hi, int(value))) def _fit(text, limit: int) -> str | None: """Fit prose into `limit` chars WITHOUT cutting mid-sentence. The llama.cpp grammar force-closes strings AT maxLength, so a cut never arrives longer than the limit — it arrives near the limit ending mid-thought. Trim back to the last sentence boundary; if no boundary survives in a reasonable prefix, return None (caller rejects → fallback). """ t = str(text).strip() if len(t) > limit: t = t[:limit] else: ends_clean = bool(t) and ( t[-1] in ".!?…" or (len(t) > 1 and t[-1] in "'\")" and t[-2] in ".!?…")) if ends_clean or len(t) < limit * 0.9: return t # complete, or short enough that no cut happened cut = max(t.rfind("."), t.rfind("!"), t.rfind("?")) if cut < limit * 0.35: return None # one run-on sentence — trimming would gut it out = t[:cut + 1] # keep a trailing quote that belongs to the sentence if cut + 1 < len(t) and t[cut + 1] in "'\"": out += t[cut + 1] return out def _clean_comic_fields(payload: dict) -> None: """Soft-validate the decorative comic fields (image_prompt for FLUX, comic_caption for the UI): clamp length and drop any that carry banned/real-company/mechanics content. Never rejects the payload — a missing field just means that part of the overlay is skipped (and no image at all means no overlay).""" for field, limit in (("image_prompt", 400), ("comic_caption", 160)): v = payload.get(field) if not isinstance(v, str) or not v.strip(): payload.pop(field, None) continue scan = ALLOWED_TOOLS.sub(" ", v) if any(pat.search(scan) for pat in _BANNED) or MECHANICS_LEAK.search(scan): trace("vald", f"dropped {field} (banned/mechanics content)") payload.pop(field, None) continue payload[field] = v[:limit] def _self_narration(npc_id: str, text: str) -> bool: """First-person dialogue must not describe the speaker from outside: ' is/was/says/looked...' is narration. A bare self-reference ('the Brad timeline', 'Brad-window') is comedy and stays legal.""" name = npc_id.capitalize() return re.search( rf"\b{name}(?:'s)? (?:is|was|will|would|has|had|says?|said|seems?|" rf"seemed|looks?|looked|takes?|took|stares?|stared|watch(?:es|ed)?|" rf"remain(?:s|ed)?|sweat(?:s|ing)?|reject(?:s|ed)|just)\b", str(text), re.IGNORECASE) is not None def validate_crisis(payload: dict, state: GameState, npc_id: str, bribe_offer: int = 0) -> dict | None: """Returns a cleaned payload, or None if it must be replaced by fallback.""" try: required = {"npc_reaction", "consequence", "revenue_delta", "animation", "boss_title", "log_entry", "morale_delta", "npc_id", "relationship_delta", "pocket_money_delta", "special_next_event"} missing = required - payload.keys() if missing: return _reject("crisis", f"missing fields {sorted(missing)}") if payload["animation"] not in ANIMATION_TRIGGERS: return _reject("crisis", f"unknown animation '{payload['animation']}'") if payload["animation"] == "confetti_burst": return _reject("crisis", "confetti_burst reserved for the win screen") if payload["animation"] == "bribery_envelope" and bribe_offer == 0: trace("vald", "remap: bribery_envelope -> npc_confused (no bribe active)") payload["animation"] = "npc_confused" # models overpick this one if payload["npc_id"] not in NPC_IDS: trace("vald", f"fix: npc_id '{payload['npc_id']}' -> {npc_id}") payload["npc_id"] = npc_id sne = payload["special_next_event"] if sne is not None and sne not in SPECIAL_EVENT_TYPES: trace("vald", f"fix: special_next_event '{sne}' -> null") payload["special_next_event"] = None word = banned_match(payload) if word: return _reject("crisis", f"banned content '{word}'") leak = MECHANICS_LEAK.search( f"{payload['npc_reaction']} {payload['consequence']} " f"{payload['log_entry']}") if leak: return _reject("crisis", f"mechanics language in prose: " f"'{leak.group(0)}'") if _self_narration(npc_id, payload["npc_reaction"]): return _reject("crisis", f"npc_reaction narrates {npc_id} " "in third person (own name in dialogue)") # romance gating if (state.npc(npc_id).relationship < 65 and ROMANCE_WORDS.search(_text_fields(payload))): return _reject("crisis", f"romance content below 65 " f"(rel={state.npc(npc_id).relationship})") # numeric clamps before = payload["revenue_delta"] payload["revenue_delta"] = _clamp(payload["revenue_delta"], -300_000, 400_000) if state.revenue + payload["revenue_delta"] < 0: payload["revenue_delta"] = -state.revenue # revenue floor if before < payload["revenue_delta"]: payload["floored_loss"] = True # internal: loss hit the floor if payload["revenue_delta"] != before: trace("vald", f"clamp: revenue_delta {before} -> {payload['revenue_delta']}") payload["morale_delta"] = _clamp(payload["morale_delta"], -25, 15) payload["relationship_delta"] = _clamp(payload["relationship_delta"], -20, 15) pm_before = payload["pocket_money_delta"] payload["pocket_money_delta"] = _clamp( payload["pocket_money_delta"], 0, max(0, bribe_offer)) if payload["pocket_money_delta"] != pm_before: trace("vald", f"clamp: pocket_money_delta {pm_before} -> " f"{payload['pocket_money_delta']} (offer={bribe_offer})") # sentence-safe length fitting + capitalize sentence starts for key, n in (("npc_reaction", 300), ("consequence", 220), ("log_entry", 150)): fitted = _fit(payload[key], n) if fitted is None: return _reject("crisis", f"{key} overruns {n} chars with no " "sentence boundary") payload[key] = fitted[:1].upper() + fitted[1:] if fitted else fitted title = str(payload["boss_title"])[:60] payload["boss_title"] = title[:1].upper() + title[1:] if title else title # the log feeds board presentations: its dollar figure MUST be the # actually-applied revenue (post-floor), never the model's pre-floor # claim — otherwise the board interrogates a loss that never landed. delta = payload["revenue_delta"] # already floored above base = re.sub( r"\s*(,?\s*(resulting in|for a|costing|netting|losing|gaining|" r"leading to)\b.*|[-+]?\$[\d,]+\s*[KkMm]?.*)$", "", str(payload["log_entry"]), flags=re.IGNORECASE).rstrip(" .,;—-") if not base: base = f"{npc_id.capitalize()} handled it" suffix = (f"{'+' if delta > 0 else '-'}${abs(delta) // 1000}K." if delta else "No revenue impact.") payload["log_entry"] = f"{base}. {suffix}"[:150] _clean_comic_fields(payload) return payload except (TypeError, ValueError, KeyError) as exc: return _reject("crisis", f"malformed payload: {exc!r}") def validate_event(payload: dict, state: GameState) -> dict | None: try: required = {"affected_npc", "category", "headline", "intro", "option_a", "option_b", "urgency", "setup_animation", "morale_preview"} missing = required - payload.keys() if missing: return _reject("event", f"missing fields {sorted(missing)}") if payload["affected_npc"] not in NPC_IDS + ["player"]: return _reject("event", f"bad affected_npc '{payload['affected_npc']}'") if payload["category"] not in EVENT_CATEGORIES: return _reject("event", f"bad category '{payload['category']}'") if payload["setup_animation"] not in ANIMATION_TRIGGERS: trace("vald", f"fix: setup_animation '{payload['setup_animation']}' " "-> npc_confused") payload["setup_animation"] = "npc_confused" word = banned_match(payload) if word: return _reject("event", f"banned content '{word}'") leak = MECHANICS_LEAK.search( f"{payload['headline']} {payload['intro']} " f"{payload['option_a']} {payload['option_b']}") if leak: return _reject("event", f"mechanics language in prose: " f"'{leak.group(0)}'") # the intro is the NPC SPEAKING — never narration about them if (payload["affected_npc"] != "player" and _self_narration(payload["affected_npc"], payload["intro"])): return _reject("event", f"intro narrates " f"{payload['affected_npc']} in third person") # dedup: headline must not fuzzy-match an existing log entry head = str(payload["headline"]).lower()[:40] if head and any(head in entry.lower() for entry in state.event_log): return _reject("event", f"duplicate of logged event: '{head}'") payload["morale_preview"] = _clamp(payload["morale_preview"], -20, 10) payload["headline"] = str(payload["headline"])[:80] for key, n in (("intro", 280), ("option_a", 160), ("option_b", 160), ("urgency", 120)): fitted = _fit(payload[key], n) if fitted is None: return _reject("event", f"{key} overruns {n} chars with no " "sentence boundary") payload[key] = fitted _clean_comic_fields(payload) return payload except (TypeError, ValueError, KeyError) as exc: return _reject("event", f"malformed payload: {exc!r}") def validate_presentation(payload: dict, state: GameState, round_no: int, closing: bool, prev_dialogues: tuple = ()) -> dict | None: import difflib try: base = {"round", "board_tone", "event_referenced", "round_difficulty", "board_dialogue"} extra = {"cumulative_score"} if closing else {"option_a", "option_b"} missing = (base | extra) - payload.keys() if missing: return _reject("pres", f"missing fields {sorted(missing)}") if int(payload["round"]) != round_no: trace("vald", f"fix: round {payload['round']} -> {round_no}") payload["round"] = round_no if payload["board_tone"] not in BOARD_TONES: return _reject("pres", f"bad board_tone '{payload['board_tone']}'") if payload["round_difficulty"] not in ROUND_DIFFICULTIES: trace("vald", f"fix: round_difficulty " f"'{payload['round_difficulty']}' -> standard") payload["round_difficulty"] = "standard" word = banned_match(payload) if word: return _reject("pres", f"banned content '{word}'") # the board must not repeat itself across rounds dialogue = str(payload["board_dialogue"]).lower() for prev in prev_dialogues: ratio = difflib.SequenceMatcher( None, dialogue, str(prev).lower()).ratio() if ratio > 0.6: return _reject("pres", f"round repeats earlier dialogue " f"(similarity {ratio:.2f})") # board options must not duplicate each other or the dialogue if not closing: a, b = str(payload["option_a"]).lower(), str(payload["option_b"]).lower() if difflib.SequenceMatcher(None, a, b).ratio() > 0.85: return _reject("pres", "option_a and option_b are the same") if len(a) > 30 and a[:60] in dialogue: return _reject("pres", "options duplicate the board dialogue") # the referenced event must actually exist in the log ref = str(payload["event_referenced"]).lower() if state.event_log and not any( ref[:30] in e.lower() or e.lower()[:30] in ref for e in state.event_log): return _reject("pres", f"event_referenced not in log: '{ref[:60]}'") if closing: payload["cumulative_score"] = _clamp(payload["cumulative_score"], 0, 100) fitted = _fit(payload["board_dialogue"], 360) if fitted is None: return _reject("pres", "board_dialogue overruns with no " "sentence boundary") payload["board_dialogue"] = fitted if not closing: for key in ("option_a", "option_b"): opt = _fit(payload[key], 160) if opt is None: return _reject("pres", f"{key} overruns with no " "sentence boundary") payload[key] = opt return payload except (TypeError, ValueError, KeyError) as exc: return _reject("pres", f"malformed payload: {exc!r}") def _prose_ok(kind: str, payload: dict, *texts: str) -> bool: """Shared banned-content + mechanics-leak gate for idle prose.""" word = banned_match(payload) if word: _reject(kind, f"banned content '{word}'") return False leak = MECHANICS_LEAK.search(" ".join(texts)) if leak: _reject(kind, f"mechanics language in prose: '{leak.group(0)}'") return False return True def validate_chat(payload: dict, state: GameState, npc_id: str) -> dict | None: try: if not {"npc_line", "relationship_delta", "morale_delta"} \ .issubset(payload): return _reject("chat", "missing fields") if not _prose_ok("chat", payload, str(payload["npc_line"])): return None if (state.npc(npc_id).relationship < 65 and ROMANCE_WORDS.search(str(payload["npc_line"]))): return _reject("chat", f"romance below 65 " f"(rel={state.npc(npc_id).relationship})") if _self_narration(npc_id, payload["npc_line"]): return _reject("chat", f"npc_line narrates {npc_id} in third person") payload["relationship_delta"] = _clamp(payload["relationship_delta"], -2, 2) payload["morale_delta"] = _clamp(payload["morale_delta"], -1, 1) text = _fit(payload["npc_line"], 160) if text is None: return _reject("chat", "npc_line overruns with no sentence boundary") payload["npc_line"] = text[:1].upper() + text[1:] if text else text return payload except (TypeError, ValueError, KeyError) as exc: return _reject("chat", f"malformed payload: {exc!r}") def validate_banter(payload: dict, state: GameState) -> dict | None: try: if not {"npc_id", "line"}.issubset(payload): return _reject("banter", "missing fields") if payload["npc_id"] not in NPC_IDS: return _reject("banter", f"bad npc_id '{payload['npc_id']}'") if not _prose_ok("banter", payload, str(payload["line"])): return None text = _fit(payload["line"], 110) if text is None: return _reject("banter", "line overruns with no sentence boundary") payload["line"] = text[:1].upper() + text[1:] if text else text return payload except (TypeError, ValueError, KeyError) as exc: return _reject("banter", f"malformed payload: {exc!r}") def validate_eavesdrop(payload: dict, state: GameState, pair: tuple[str, str]) -> dict | None: try: lines = payload.get("lines") if not isinstance(lines, list) or not 2 <= len(lines) <= 3: return _reject("eavs", "lines must be a list of 2-3 entries") all_text = [] for entry in lines: if entry.get("speaker") not in pair: return _reject("eavs", f"speaker '{entry.get('speaker')}' " f"not in pair {pair}") text = _fit(entry.get("line", ""), 110) if text is None: return _reject("eavs", "a line overruns with no sentence boundary") entry["line"] = text[:1].upper() + text[1:] if text else text all_text.append(entry["line"]) if not _prose_ok("eavs", payload, *all_text): return None return payload except (TypeError, ValueError, KeyError) as exc: return _reject("eavs", f"malformed payload: {exc!r}") def validate_email(payload: dict, state: GameState) -> dict | None: try: if not {"sender", "subject", "body"}.issubset(payload): return _reject("mail", "missing fields") if payload["sender"] not in NPC_IDS + ["system"]: return _reject("mail", f"bad sender '{payload['sender']}'") if not _prose_ok("mail", payload, str(payload["subject"]), str(payload["body"])): return None subject = str(payload["subject"])[:60] payload["subject"] = subject[:1].upper() + subject[1:] if subject else subject body = _fit(payload["body"], 240) if body is None: return _reject("mail", "body overruns with no sentence boundary") payload["body"] = body[:1].upper() + body[1:] if body else body return payload except (TypeError, ValueError, KeyError) as exc: return _reject("mail", f"malformed payload: {exc!r}") def validate_verdict(payload: dict) -> dict | None: try: if "verdict" not in payload: return _reject("verd", "missing verdict field") word = banned_match(payload) if word: return _reject("verd", f"banned content '{word}'") payload["verdict"] = str(payload["verdict"])[:300] return payload except (TypeError, ValueError) as exc: return _reject("verd", f"malformed payload: {exc!r}")