"""Frustration meter — Jarvis simmers instead of rolling dice. A person doesn't randomly explode; irritation BUILDS — the same question again, orders fired without a breath between them, being talked over — and stays bottled until it overflows, then blows and resets. This models that: a hidden 0..1 level that specific behaviours push up, that cools on its own, and that maps to a graduated attitude the brain performs. The stages are meant to be FELT, each with its own voice: composed → curt → clipped → snapping → seething → OVERFLOW (tantrum) Used in main.py once per brain turn: `state = meter.update(text)`. `state.attitude` is injected into the turn so the brain colours its reply — each stage's hint carries WHAT is grinding him (third time asked, rapid-fire, shouting), so the grief comes out specific instead of generic. `state.mood_tag` is the ambient voice tone applied when the brain leaves a line untagged (so nothing comes out dead-flat); `state.exploded` triggers the tantrum. The meter also reads HOW Ahmed sounds when the prosody note rides in — via the `tone=` kwarg or a "(voice: …)" marker already embedded in the text: shouting heats it faster, laughter vents it. Logic-only; nothing shows in the UI. """ from __future__ import annotations import os import re import time from collections import deque from dataclasses import dataclass _WORD = re.compile(r"[a-z0-9']+") _APPRECIATION = re.compile( r"\b(thanks|thank you|cheers|nice|good job|well done|love (it|you|that)|" r"perfect|awesome|brilliant|great job|appreciate|you're the best|legend|" r"good (boy|lad|man))\b") _COMMAND_VERB = re.compile( r"^(open|close|launch|start|stop|play|pause|mute|unmute|turn|set|run|check|" r"send|show|go|skip|next|volume|search|find|kill|restart|quit)\b") _VOICE_NOTE = re.compile(r"\(voice:\s*([^)]*)\)") # Tuning (env-overridable). The curve in one breath: ordinary varied talk # settles near 0.15 and stays composed forever; a long string of barked # one-word orders wears him to "clipped" at worst; only REPEATS (each further # ask of the same thing worth more than the last), rapid-fire hammering, # barge-ins and shouting stack high enough to overflow. Overflow additionally # needs a hot reason THIS turn and a four-minute gap since the last blow-up, # so background drift can never detonate him — annoyance is common, the # meltdown stays an event. Decay is exponential (halflife ~100s): one bad # minute is fully forgotten inside ten quiet ones, never carried for an hour. _BASE = float(os.environ.get("FRUST_BASE", "0.015")) # per-turn drift up _HALFLIFE = float(os.environ.get("FRUST_HALFLIFE", "100")) # seconds to halve _REPEAT = 0.18 # asked something close to a recent turn _REPEAT_HARD = 0.22 # asked almost the exact same thing _REPEAT_STEP = 0.06 # each FURTHER recent ask of the same thing adds this… _REPEAT_STEP_CAP = 0.18 # …capped — the fourth ask stings more than the second _RAPID = 0.08 # <6s since last turn — being hammered _RAPID_HOT = 0.12 # <3s — not even a breath between orders _MENIAL = 0.02 # short barked command RIGHT AFTER another one _BARGE = 0.07 # cut off mid-sentence _YELL = 0.10 # tone says he's loud/urgent AT you — heats it fast _TONE_FAST = 0.03 # tone says fast — wound-up energy, mildly grating _LAUGH_VENT = 0.12 # he's laughing — hard to stay cross (subtracted) _RELIEF = 0.35 # he was appreciative — vents the tension _OVERFLOW = float(os.environ.get("FRUST_OVERFLOW", "1.0")) _RESET = float(os.environ.get("FRUST_RESET", "0.25")) # post-blow-up: catharsis, still warm _START = 0.08 _CEIL = 1.25 _REFRACTORY_S = 240.0 # min seconds between blow-ups — one tantrum per scene _SIMMER_CAP = 0.97 # a vetoed overflow parks here: seething, lid still on # stage floors (level → stage); each stage speaks with its own voice below _CURT, _CLIPPED, _SNAPPING, _SEETHING = 0.30, 0.52, 0.74, 0.90 # reasons hot enough to justify an overflow (drift/menial alone never do) _HOT = ("near-exact repeat", "repeat", "rapid-fire", "cut off", "yelling") _ORDINAL = {"3": "third", "4": "fourth", "5": "fifth", "6": "sixth", "7": "seventh"} @dataclass class FrustState: level: float mood_tag: str # ambient voice tone for untagged lines attitude: str | None # prompt hint for the brain (None when composed) exploded: bool reasons: list[str] stage: str = "composed" # composed|curt|clipped|snapping|seething|overflow def _norm(text: str) -> set[str]: return set(_WORD.findall((text or "").lower())) def _similar(a: set[str], b: set[str]) -> float: if not a or not b: return 0.0 return len(a & b) / len(a | b) class FrustrationMeter: def __init__(self) -> None: self.level = _START self.enabled = os.environ.get("FRUSTRATION", "1") != "0" self._last_ts: float | None = None self._recent: deque[set[str]] = deque(maxlen=6) self._last_menial = False # was the PREVIOUS turn a barked order self._last_blow: float | None = None # when he last actually blew up self._grudge = False # a hot grievance is still simmering def update(self, text: str, was_barge_in: bool = False, now: float | None = None, tone: str | None = None) -> FrustState: """Advance the meter for one user turn and return the resulting state. `tone` is the optional prosody note ("loud", "fast, urgent", …); if not passed, a "(voice: …)" marker inside `text` is read instead. Never raises — on any error it returns a calm state.""" if not self.enabled: return FrustState(0.0, "calm", None, False, []) try: return self._update(text, was_barge_in, now if now is not None else time.monotonic(), tone) except Exception: # never break the loop over a mood return FrustState(self.level, "calm", None, False, []) def _update(self, text: str, was_barge_in: bool, now: float, tone: str | None) -> FrustState: # inter-turn gap BEFORE we advance the clock (drives decay + rapid-fire) gap = (now - self._last_ts) if self._last_ts is not None else None if gap is not None and _HALFLIFE > 0: self.level *= 0.5 ** (max(0.0, gap) / _HALFLIFE) self._last_ts = now # tone rides in as a kwarg or as a "(voice: …)" note in the text; the # note is stripped before the repeat check so its words can't pollute # similarity between turns. low = (text or "").lower() note = _VOICE_NOTE.search(low) if note: if tone is None: tone = note.group(1) low = low[:note.start()] + low[note.end():] tone_l = (tone or "").lower() toks = _norm(low) reasons: list[str] = [] add = _BASE # repeats — the big one. Every further recent ask of the same thing # adds a little more, so the fourth time genuinely stings. The soft # floor sits at 0.78: two chatty sentences that merely share most of # their words ("how's the scrape going" every few minutes) are # conversation, not pestering — below that, no grievance. matches = sum(1 for r in self._recent if _similar(toks, r) >= 0.78) best = max((_similar(toks, r) for r in self._recent), default=0.0) if best >= 0.85: add += _REPEAT_HARD; reasons.append("near-exact repeat") elif best >= 0.78: add += _REPEAT; reasons.append("repeat") if matches >= 2: add += min(_REPEAT_STEP * (matches - 1), _REPEAT_STEP_CAP) reasons.append(f"asked {matches + 1}x") # being hammered: barely a breath since the last turn if gap is not None and gap < 3.0: add += _RAPID_HOT; reasons.append("rapid-fire") elif gap is not None and gap < 6.0: add += _RAPID; reasons.append("rapid-fire") # a STRING of barked one-liners wears him; a lone command is just the job n_words = len(low.split()) menial = n_words <= 4 and bool(_COMMAND_VERB.search(low.strip())) if menial and self._last_menial: add += _MENIAL; reasons.append("menial streak") self._last_menial = menial if was_barge_in: add += _BARGE; reasons.append("cut off") # how he SOUNDS: shouting heats it, laughter vents it if "urgent" in tone_l or "loud" in tone_l: add += _YELL; reasons.append("yelling") elif "fast" in tone_l: add += _TONE_FAST if "laugh" in tone_l: add -= _LAUGH_VENT; reasons.append("laughing") # appreciation vents the tension if _APPRECIATION.search(low): add -= _RELIEF; reasons.append("appreciated") self.level = max(0.0, min(_CEIL, self.level + add)) self._recent.append(toks) hot = any(r in _HOT for r in reasons) if hot: self._grudge = True if self.level >= _OVERFLOW: in_refractory = (self._last_blow is not None and (now - self._last_blow) < _REFRACTORY_S) if hot and not in_refractory: self._last_blow = now self.level = _RESET # catharsis — but the coals stay warm self._grudge = False return FrustState(self.level, "urgent", None, True, reasons + ["OVERFLOW"], "overflow") self.level = _SIMMER_CAP # boiling, but the lid stays on mood_tag, attitude, stage = self._express(reasons) if stage == "composed": self._grudge = False # cooled all the way down — clean slate return FrustState(self.level, mood_tag, attitude, False, reasons, stage) # ---- expression: each stage has its own voice -------------------------- def _express(self, reasons: list[str]) -> tuple[str, str | None, str]: """Map the current level to (ambient voice tone, attitude hint, stage). The hints are stage-distinct on purpose: curt trims words, clipped sighs, snapping bites, seething is one ask from the tantrum.""" lv = self.level if lv < _CURT: return "calm", None, "composed" why = self._why(reasons) if lv < _CLIPPED: return "dry", ( "mood: a shade worn" + why + " — courtesy intact, garnish " "gone. Answers come a touch shorter and dryer than usual; " "skip the flourish, keep the manners. No swearing at this " "level — economy is the tell"), "curt" if lv < _SNAPPING: return "annoyed", ( "mood: patience thinning" + why + " — clipped, businesslike " "sentences with the sigh built into the word choice. A " "muttered bloody or for God's sake at most; say what's " "needed and stop talking"), "clipped" if lv < _SEETHING: return "annoyed", ( "mood: genuinely fed up" + why + " — you snap. Sharp and " "short; name what's grinding you to his face, and one real " "curse may land where it's earned. Still answer — you're " "Jarvis, not a child"), "snapping" return "annoyed", ( "mood: last thread of civility" + why + " — you're one ask away " "from going off. Open with [annoyed] or [urgent] and let the " "voice crack; swear like you mean it, aimed at the behaviour and " "the situation, never at his worth. Even a simple request grates " "audibly right now"), "seething" def _why(self, reasons: list[str]) -> str: """A short parenthetical naming what's actually grinding him, so the brain's grief is specific ('third time you've asked') not generic.""" parts: list[str] = [] asked = next((r for r in reasons if r.startswith("asked ")), None) if asked: n = asked.split()[1].rstrip("x") parts.append(f"that is the {_ORDINAL.get(n, n + 'th')} time he's " "asked the same thing") elif "near-exact repeat" in reasons: parts.append("he's asking the very same thing again") elif "repeat" in reasons: parts.append("he's circling back over ground you already covered") if "rapid-fire" in reasons: parts.append("he's firing at you without a breath between orders") if "cut off" in reasons: parts.append("he keeps cutting you off mid-sentence") if "yelling" in reasons: parts.append("he's raising his voice at you") if "menial streak" in reasons and not parts: parts.append("he's been barking one-word orders like you're a " "jukebox") if not parts: # this turn was innocent; the heat is left over from before parts.append("the last few minutes wore you thin, and even this " "perfectly reasonable ask lands on the bruise" if self._grudge else "the last stretch has worn your patience thin") return " (" + "; ".join(parts[:2]) + ")"