# This file is the core “director + runtime engine” of the puppet theater system. # It controls story progression, selects speakers, validates AI decisions, # runs one beat at a time, and manages both deterministic and LLM-based directors. from collections import Counter from dataclasses import dataclass import json import time from pydantic import ValidationError from puppet_theater.backends import ( DEFAULT_HF_API_MODEL_ID, DEFAULT_OPENBMB_MODEL_ID, HFAPIBackend, HF_API_DIRECTOR_MAX_TOKENS, HF_API_DIRECTOR_TEMPERATURE, OpenBMBTransformersBackend, generate_actor_response, get_backend, ) from puppet_theater.models import Actor, Beat, DirectorDecision, TheaterSession from puppet_theater.prompts import DIRECTOR_DECISION_PROMPT from puppet_theater.trace import add_trace_event from puppet_theater.tools import run_actor_tool_request # Allowed narrative beats (types of actions that can happen in the story) BEAT_ARC = [ "setup", "denial_or_contradiction", "evidence_or_prop", "secret_reveal", "chaos_or_intervention", "finale", ] # High-level story phases derived from progress STORY_PHASES = ["opening", "complication", "reveal", "chaos", "finale"] def story_progress(session: TheaterSession) -> float: # Returns normalized progress (0.0 to 1.0) based on beat progression target_beats = max(1, session.target_beats) return min(1.0, max(0.0, session.beat_index / target_beats)) def story_phase(session: TheaterSession) -> str: # Converts numeric progress into narrative phase buckets progress = story_progress(session) if progress < 0.20: return "opening" if progress < 0.45: return "complication" if progress < 0.65: return "reveal" if progress < 0.85: return "chaos" return "finale" def should_force_finale(session: TheaterSession) -> bool: # Determines if story should end early due to constraints or user request return ( session.finale_requested or session.beat_index >= session.target_beats - 1 or session.beat_index >= session.max_beats - 1 ) @dataclass(frozen=True) class DirectorGeneration: # Wrapper around a director decision plus metadata about how it was generated decision: DirectorDecision director_mode: str model_id: str | None validation_status: str fallback_used: bool reason_summary: str latency_ms: int | None = None error: str | None = None class DirectorPolicy: # Deterministic rule-based director (fallback when no AI is used) def decide(self, session: TheaterSession) -> DirectorDecision: phase = story_phase(session) progress = story_progress(session) # Force finale conditions override everything else if should_force_finale(session): speaker = self._choose_speaker(session) return DirectorDecision( next_speaker=speaker.name, beat_type="finale", instruction="Tie the scene together in one crisp curtain-call line.", stage_effect="curtain_fall", should_end_scene=True, reason_summary="Finale requested, target reached, or final beat reached; closing cleanly.", ) # Audience prop takes priority in narrative if session.latest_prop is not None: speaker = self._choose_speaker(session) return DirectorDecision( next_speaker=speaker.name, beat_type="evidence_or_prop", instruction=f"Use the {session.latest_prop} as evidence and make it matter immediately.", stage_effect="prop_table_glow", uses_prop=True, reason_summary=f"Latest audience prop {session.latest_prop} should affect the next beat.", ) # Newly summoned actors should get priority spotlight summoned_actor = self._unused_summoned_actor(session) if summoned_actor is not None: return DirectorDecision( next_speaker=summoned_actor.name, beat_type="chaos_or_intervention", instruction="Enter the scene with one specific complication, then hand control back.", stage_effect="entrance_sparkle", reason_summary=f"Newly summoned actor {summoned_actor.name} has not spoken yet.", ) # Default storytelling logic based on phase beat_type = self._beat_type_for_phase(phase, session) speaker = self._choose_speaker(session) reveal_secret = beat_type == "secret_reveal" return DirectorDecision( next_speaker=speaker.name, beat_type=beat_type, instruction=self._instruction_for_beat(beat_type, reveal_secret), stage_effect=self._effect_for_beat(beat_type), reveal_secret=reveal_secret, should_end_scene=beat_type == "finale", reason_summary=f"Following {phase} phase at {progress:.0%} progress with {beat_type}.", ) def _choose_speaker(self, session: TheaterSession) -> Actor: # Picks next actor fairly based on usage + rotation latest_speaker = session.transcript[-1].speaker if session.transcript else None counts = Counter(beat.speaker for beat in session.transcript) candidates = [actor for actor in session.actors if actor.name != latest_speaker] or session.actors target_index = session.beat_index % len(session.actors) indexed_actor = session.actors[target_index] if indexed_actor in candidates and counts[indexed_actor.name] <= min(counts[actor.name] for actor in candidates): return indexed_actor return min(candidates, key=lambda actor: (counts[actor.name], session.actors.index(actor))) def _unused_summoned_actor(self, session: TheaterSession) -> Actor | None: # Returns summoned actors who haven't spoken yet spoken = {beat.speaker for beat in session.transcript} for actor in session.actors[3:]: if actor.name not in spoken: return actor return None def _instruction_for_beat(self, beat_type: str, reveal_secret: bool) -> str: # Maps beat type to actor instruction if reveal_secret: return "Reveal your secret directly, but keep it playful and stage-ready." return { "setup": "Establish the premise and your role in one clear line.", "denial_or_contradiction": "Contradict the premise with theatrical confidence.", "evidence_or_prop": "Present a clue or prop as if it changes the whole case.", "chaos_or_intervention": "React to the audience and escalate the scene briefly.", "finale": "End the scene with a clean button and a bow.", }.get(beat_type, "Keep the scene moving with one short line.") def _effect_for_beat(self, beat_type: str) -> str: # Maps beat type to stage visual effect return { "setup": "warm_spotlight", "denial_or_contradiction": "quick_blackout", "evidence_or_prop": "prop_table_glow", "secret_reveal": "single_spotlight", "chaos_or_intervention": "confetti_rustle", "finale": "curtain_fall", }[beat_type] def _beat_type_for_phase(self, phase: str, session: TheaterSession) -> str: # Chooses beat type based on story phase progression if phase == "opening": return "setup" if session.beat_index == 0 else "denial_or_contradiction" if phase == "complication": return "evidence_or_prop" if session.beat_index % 2 == 0 else "denial_or_contradiction" if phase == "reveal": return "secret_reveal" if phase == "chaos": return "chaos_or_intervention" return "chaos_or_intervention" class OpenBMBDirectorPolicy: # Director powered by OpenBMB transformer backend def __init__(self, fallback_policy: DirectorPolicy | None = None) -> None: self.fallback_policy = fallback_policy or DirectorPolicy() def decide(self, session: TheaterSession) -> DirectorGeneration: start_time = time.perf_counter() backend = get_backend( "openbmb", max_new_tokens=session.backend_max_new_tokens, temperature=session.backend_temperature, ) if not isinstance(backend, OpenBMBTransformersBackend): return self._fallback(session, "backend_unavailable", start_time) prompt = build_director_prompt(session) try: raw_output = backend._generate_text(prompt) except Exception as exc: return self._fallback(session, "backend_error", start_time, _summarize_error(exc), backend.model_id) decision, validation_status = parse_director_decision(raw_output, session) if decision is not None: return DirectorGeneration( decision=decision, director_mode="openbmb", model_id=backend.model_id, validation_status=validation_status, fallback_used=False, reason_summary=decision.reason_summary, latency_ms=_elapsed_ms(start_time), ) # Repair attempt when output is invalid repair_prompt = ( f"{prompt}\n\nPrevious output failed validation with status {validation_status}.\n" "Return only valid JSON matching the required schema. No markdown. No extra keys.\n" f"Previous output: {raw_output}" ) try: repair_output = backend._generate_text(repair_prompt) except Exception as exc: return self._fallback( session, f"{validation_status};repair_backend_error", start_time, _summarize_error(exc), backend.model_id, ) decision, repair_status = parse_director_decision(repair_output, session) if decision is not None: return DirectorGeneration( decision=decision, director_mode="openbmb", model_id=backend.model_id, validation_status=f"repair_{repair_status}", fallback_used=False, reason_summary=decision.reason_summary, latency_ms=_elapsed_ms(start_time), ) return self._fallback( session, f"{validation_status};repair_{repair_status}", start_time, model_id=backend.model_id, ) def _fallback( self, session: TheaterSession, validation_status: str, start_time: float, error: str | None = None, model_id: str | None = None, ) -> DirectorGeneration: decision = self.fallback_policy.decide(session) return DirectorGeneration( decision=decision, director_mode="openbmb", model_id=model_id or DEFAULT_OPENBMB_MODEL_ID, validation_status=validation_status, fallback_used=True, reason_summary=decision.reason_summary, latency_ms=_elapsed_ms(start_time), error=error, ) class HFAPIDirectorPolicy: # Director using HuggingFace API backend def __init__(self, fallback_policy: DirectorPolicy | None = None) -> None: self.fallback_policy = fallback_policy or DirectorPolicy() def decide(self, session: TheaterSession) -> DirectorGeneration: start_time = time.perf_counter() backend = get_backend( "hf_api", max_new_tokens=HF_API_DIRECTOR_MAX_TOKENS, temperature=HF_API_DIRECTOR_TEMPERATURE, ) if not isinstance(backend, HFAPIBackend): return self._fallback(session, "backend_unavailable", start_time) prompt = build_director_prompt(session) try: raw_output = backend._generate_text( prompt, max_tokens=HF_API_DIRECTOR_MAX_TOKENS, temperature=HF_API_DIRECTOR_TEMPERATURE, ) except Exception as exc: generation = self._fallback(session, "backend_error", start_time, _summarize_error(exc), backend.model_id) _record_director_backend_status(backend, generation) return generation decision, validation_status = parse_director_decision(raw_output, session) if decision is not None: generation = DirectorGeneration( decision=decision, director_mode="hf_api", model_id=backend.model_id, validation_status=validation_status, fallback_used=False, reason_summary=decision.reason_summary, latency_ms=_elapsed_ms(start_time), ) _record_director_backend_status(backend, generation) return generation repair_prompt = ( f"{prompt}\n\nPrevious output failed validation with status {validation_status}.\n" "Return only valid JSON matching the required schema. No markdown. No extra keys.\n" f"Previous output: {raw_output}" ) try: repair_output = backend._generate_text( repair_prompt, max_tokens=HF_API_DIRECTOR_MAX_TOKENS, temperature=HF_API_DIRECTOR_TEMPERATURE, ) except Exception as exc: generation = self._fallback( session, f"{validation_status};repair_backend_error", start_time, _summarize_error(exc), backend.model_id, ) _record_director_backend_status(backend, generation) return generation decision, repair_status = parse_director_decision(repair_output, session) if decision is not None: generation = DirectorGeneration( decision=decision, director_mode="hf_api", model_id=backend.model_id, validation_status=f"repair_{repair_status}", fallback_used=False, reason_summary=decision.reason_summary, latency_ms=_elapsed_ms(start_time), ) _record_director_backend_status(backend, generation) return generation generation = self._fallback( session, f"{validation_status};repair_{repair_status}", start_time, model_id=backend.model_id, ) _record_director_backend_status(backend, generation) return generation def _fallback( self, session: TheaterSession, validation_status: str, start_time: float, error: str | None = None, model_id: str | None = None, ) -> DirectorGeneration: decision = self.fallback_policy.decide(session) return DirectorGeneration( decision=decision, director_mode="hf_api", model_id=model_id or DEFAULT_HF_API_MODEL_ID, validation_status=validation_status, fallback_used=True, reason_summary=decision.reason_summary, latency_ms=_elapsed_ms(start_time), error=error, ) def choose_director_decision(session: TheaterSession) -> DirectorGeneration: # Router that selects deterministic vs AI directors if session.director_mode == "hf_api": return HFAPIDirectorPolicy().decide(session) if session.director_mode == "openbmb": return OpenBMBDirectorPolicy().decide(session) decision = DirectorPolicy().decide(session) return DirectorGeneration( decision=decision, director_mode="deterministic", model_id=None, validation_status="valid", fallback_used=False, reason_summary=decision.reason_summary, latency_ms=0, ) def run_one_beat(session: TheaterSession | None) -> TheaterSession | None: # Executes one full story cycle: director → actor → state update → logs if session is None: return None if session.beat_index >= session.max_beats: session.director_log.append("Curtain already fallen; no new beat added.") add_trace_event( session, "beat_skipped", validation_status="curtain_already_fallen", fallback_used=False, ) return session current_phase = story_phase(session) current_progress = story_progress(session) director_generation = choose_director_decision(session) decision = director_generation.decision speaker = _actor_by_name(session, decision.next_speaker) prop = session.latest_prop if decision.uses_prop else None if prop is not None: speaker.held_prop = prop session.director_log.append( "Director decision: " f"{decision.beat_type} for {speaker.name}; {decision.reason_summary}" ) session.director_log.append( f"Story phase {current_phase} at {current_progress:.0%} progress " f"(min={session.min_beats}, target={session.target_beats}, max={session.max_beats})." ) session.director_log.append( "Director mode " f"{director_generation.director_mode} " f"({director_generation.validation_status}, fallback={director_generation.fallback_used}, " f"latency={director_generation.latency_ms}ms)." ) if director_generation.model_id: session.director_log.append(f"Director model id: {director_generation.model_id}.") if director_generation.error: session.director_log.append(f"Director fallback reason: {director_generation.error}.") add_trace_event( session, "director_decision", speaker=speaker.name, beat_type=decision.beat_type, backend_name=director_generation.director_mode, model_id=director_generation.model_id, latency_ms=director_generation.latency_ms, validation_status=director_generation.validation_status, fallback_used=director_generation.fallback_used, fallback_reason=director_generation.error, reason_summary=director_generation.reason_summary, story_phase=current_phase, progress=round(current_progress, 3), min_beats=session.min_beats, target_beats=session.target_beats, max_beats=session.max_beats, uses_prop=decision.uses_prop, reveal_secret=decision.reveal_secret, should_end_scene=decision.should_end_scene, ) if director_generation.error: add_trace_event( session, "director_error", backend_name=director_generation.director_mode, model_id=director_generation.model_id, validation_status=director_generation.validation_status, fallback_used=director_generation.fallback_used, fallback_reason=director_generation.error, ) if director_generation.fallback_used: add_trace_event( session, "director_fallback", speaker=speaker.name, backend_name=director_generation.director_mode, model_id=director_generation.model_id, validation_status=director_generation.validation_status, fallback_used=True, fallback_reason=director_generation.error or director_generation.validation_status, story_phase=current_phase, ) backend_generation = generate_actor_response(session, decision, speaker, prop) session.beat_index += 1 response = backend_generation.response beat = Beat( speaker=speaker.name, intent=response.intent, line=response.line, emotion=response.emotion, gesture=response.gesture, stage_effect=response.stage_effect or decision.stage_effect, memory_update=response.memory_update, tool_request=response.tool_request, ) session.transcript.append(beat) state_update = apply_actor_state_update(session, speaker, response, decision, prop) tool_result = run_actor_tool_request(session, speaker, response.tool_request) if tool_result is not None and tool_result.stage_effect: beat.stage_effect = tool_result.stage_effect if prop is not None: session.latest_prop = None session.director_log.append(f"{speaker.name} picked up {prop} and used it in the scene.") add_trace_event(session, "prop_used", speaker=speaker.name, prop=prop) session.director_log.append( f"Beat {session.beat_index}/{session.max_beats}: {decision.beat_type} assigned to {speaker.name}." ) session.director_log.append( "Backend " f"{backend_generation.backend_name} returned actor output " f"({backend_generation.validation_status}, fallback={backend_generation.fallback_used}" f", load_status={backend_generation.load_status}, latency={backend_generation.latency_ms}ms)." ) if backend_generation.model_id: session.director_log.append(f"Model id: {backend_generation.model_id}.") if backend_generation.error: session.director_log.append(f"Backend fallback reason: {backend_generation.error}.") add_trace_event( session, "beat_added", beat_index=session.beat_index, beat_type=decision.beat_type, speaker=speaker.name, stage_effect=beat.stage_effect, story_phase=story_phase(session), progress=round(story_progress(session), 3), ) add_trace_event( session, "actor_response", beat_index=session.beat_index, speaker=speaker.name, backend_name=backend_generation.backend_name, model_id=backend_generation.model_id, latency_ms=backend_generation.latency_ms, validation_status=backend_generation.validation_status, fallback_used=backend_generation.fallback_used, fallback_reason=backend_generation.error, load_status=backend_generation.load_status, ) add_trace_event( session, "actor_intent", beat_index=session.beat_index, speaker=speaker.name, intent=_public_actor_text(response.intent, speaker), validation_status=backend_generation.validation_status, fallback_used=backend_generation.fallback_used, ) if state_update["memory_update"]: add_trace_event( session, "actor_memory_update", beat_index=session.beat_index, speaker=speaker.name, memory_update=state_update["memory_update"], ) add_trace_event( session, "actor_state_update", beat_index=session.beat_index, speaker=speaker.name, mood=speaker.mood, current_goal=_public_actor_text(speaker.current_goal or "", speaker), goal_progress=_public_actor_text(speaker.goal_progress, speaker), held_props=list(speaker.held_props), secret_status=speaker.secret_status, memory_count=len(speaker.recent_memory), ) if backend_generation.error: add_trace_event( session, "backend_error", beat_index=session.beat_index, speaker=speaker.name, backend_name=backend_generation.backend_name, model_id=backend_generation.model_id, validation_status=backend_generation.validation_status, fallback_used=backend_generation.fallback_used, fallback_reason=backend_generation.error, ) if backend_generation.fallback_used: add_trace_event( session, "actor_fallback", beat_index=session.beat_index, speaker=speaker.name, backend_name=backend_generation.backend_name, model_id=backend_generation.model_id, validation_status=backend_generation.validation_status, fallback_used=True, fallback_reason=backend_generation.error or backend_generation.validation_status, load_status=backend_generation.load_status, ) if decision.should_end_scene or decision.beat_type == "finale": session.beat_index = session.max_beats session.finale_requested = True session.director_log.append("Finale reached; curtain falls cleanly.") add_trace_event( session, "scene_completed", beat_index=session.beat_index, story_phase="finale", reason_summary="Finale reached; curtain falls cleanly.", ) return session def run_full_act(session: TheaterSession | None) -> TheaterSession | None: # Runs the full play until completion if session is None: return None while session.beat_index < session.max_beats: run_one_beat(session) return session def _actor_by_name(session: TheaterSession, actor_name: str) -> Actor: # Lookup actor safely, fallback to rotation if missing for actor in session.actors: if actor.name == actor_name: return actor return session.actors[session.beat_index % len(session.actors)] def apply_actor_state_update( session: TheaterSession, speaker: Actor, response: object, decision: DirectorDecision, prop: str | None, ) -> dict[str, str]: # Updates actor state: mood, memory, secret status, goals intent = _short_public_text(getattr(response, "intent", "") or "Respond to the Director's cue.", speaker, 90) memory_update = _short_public_text(getattr(response, "memory_update", "") or "", speaker, 140) emotion = _short_public_text(getattr(response, "emotion", "") or "focused", speaker, 40) speaker.mood = emotion speaker.current_goal = intent speaker.goal_progress = f"Beat {session.beat_index}: {intent}" if prop is not None and prop not in speaker.held_props: speaker.held_props.append(prop) if prop is not None: speaker.held_prop = prop if decision.reveal_secret: speaker.secret_status = "revealed" elif decision.beat_type == "secret_reveal" and speaker.secret_status == "hidden": speaker.secret_status = "hinted" elif decision.beat_type == "finale" and speaker.secret_status in {"hinted", "revealed"}: speaker.secret_status = "resolved" if memory_update: speaker.recent_memory.append(memory_update) speaker.recent_memory = speaker.recent_memory[-4:] return {"intent": intent, "memory_update": memory_update, "mood": emotion} def build_director_prompt(session: TheaterSession) -> str: # Builds LLM prompt with full story context actor_profiles = "\n".join( "- " f"{actor.name}: goal={actor.goal}; style={actor.speaking_style}; " f"tools={', '.join(actor.tools) or 'none'}; secret={actor.secret}" for actor in session.actors ) recent_transcript = "\n".join( f"{beat.speaker}: {beat.line}" for beat in session.transcript[-4:] ) or "No lines yet." recent_tool_results = "\n".join( f"- {result.actor_name} used {result.tool_name}: {result.result}" for result in session.recent_tool_results[-3:] ) or "None" summoned_actors = ", ".join(actor.name for actor in session.actors[3:]) or "None" return DIRECTOR_DECISION_PROMPT.format( show_title=session.show_title, premise=session.premise, setting=session.setting, beat_index=session.beat_index, min_beats=session.min_beats, target_beats=session.target_beats, max_beats=session.max_beats, current_progress=f"{story_progress(session):.0%}", story_phase=story_phase(session), allowed_beat_types=", ".join(BEAT_ARC), actor_profiles=actor_profiles, recent_transcript=recent_transcript, recent_tool_results=recent_tool_results, audience_action=session.latest_audience_action or "None", props=", ".join(session.props) or "None", latest_prop=session.latest_prop or "None", summoned_actors=summoned_actors, finale_requested=session.finale_requested, json_schema=json.dumps( { "next_speaker": "one exact actor name from Available actors", "beat_type": "one allowed beat type", "instruction": "non-empty director instruction, 240 chars max", "stage_effect": "non-empty concise stage effect label", "uses_prop": "boolean", "reveal_secret": "boolean", "should_end_scene": "boolean", "reason_summary": "non-empty validation-safe summary, 240 chars max", }, indent=2, ), ) def parse_director_decision(raw_output: object, session: TheaterSession) -> tuple[DirectorDecision | None, str]: parsed = _coerce_director_output(raw_output) if parsed is None: return None, "invalid_schema" if "speaker_name" in parsed and "next_speaker" not in parsed: parsed["next_speaker"] = parsed["speaker_name"] try: decision = DirectorDecision.model_validate(parsed) except ValidationError: return None, "invalid_required_fields" return validate_director_decision(decision, session) def validate_director_decision(decision: DirectorDecision, session: TheaterSession) -> tuple[DirectorDecision | None, str]: actor_names = {actor.name for actor in session.actors} if decision.next_speaker not in actor_names: return None, "invalid_speaker" if decision.beat_type not in BEAT_ARC: return None, "invalid_beat_type" if len(decision.instruction) > 240 or len(decision.reason_summary) > 240: return None, "invalid_too_long" update: dict[str, object] = {} if should_force_finale(session): update["beat_type"] = "finale" update["should_end_scene"] = True update["stage_effect"] = "curtain_fall" else: if decision.should_end_scene and session.beat_index < session.target_beats - 1: update["should_end_scene"] = False if decision.beat_type == "finale" and session.beat_index < session.target_beats - 1: update["beat_type"] = _safe_non_finale_beat_type(session) update["should_end_scene"] = False update["stage_effect"] = DirectorPolicy()._effect_for_beat(str(update["beat_type"])) if decision.uses_prop and session.latest_prop is None: update["uses_prop"] = False if decision.beat_type == "secret_reveal": update["reveal_secret"] = True if update: decision = decision.model_copy(update=update) return decision, "valid" def _safe_non_finale_beat_type(session: TheaterSession) -> str: phase = story_phase(session) if phase == "opening": return "setup" if phase == "complication": return "evidence_or_prop" if phase == "reveal": return "secret_reveal" return "chaos_or_intervention" def _coerce_director_output(raw_output: object) -> dict[str, object] | None: if isinstance(raw_output, DirectorDecision): return raw_output.model_dump() if isinstance(raw_output, dict): return raw_output if isinstance(raw_output, str): text = raw_output.strip() if not text: return None if text.startswith("```"): text = text.strip("`") if "\n" in text: text = text.split("\n", maxsplit=1)[1] start = text.find("{") end = text.rfind("}") if start != -1 and end != -1 and end > start: text = text[start : end + 1] try: decoded = json.loads(text) except json.JSONDecodeError: return None return decoded if isinstance(decoded, dict) else None return None def _elapsed_ms(start_time: float) -> int: return round((time.perf_counter() - start_time) * 1000) def _summarize_error(exc: Exception) -> str: message = " ".join(str(exc).split()) or exc.__class__.__name__ return message[:180] def _trace_text(value: str) -> str: return " ".join(value.split()).replace(":", "-")[:140] def _public_actor_text(value: str, actor: Actor) -> str: text = " ".join(value.split()) if actor.secret and actor.secret_status not in {"revealed", "resolved"}: text = text.replace(actor.secret, "[redacted secret]") return text[:140] def _short_public_text(value: str, actor: Actor, max_chars: int) -> str: text = _public_actor_text(value, actor) if len(text) <= max_chars: return text return text[: max_chars - 3].rstrip() + "..." def _record_director_backend_status(backend: object, generation: DirectorGeneration) -> None: if hasattr(backend, "latest_latency_ms"): backend.latest_latency_ms = generation.latency_ms if hasattr(backend, "latest_validation_status"): backend.latest_validation_status = generation.validation_status if hasattr(backend, "latest_fallback_used"): backend.latest_fallback_used = generation.fallback_used if hasattr(backend, "latest_fallback_reason"): backend.latest_fallback_reason = generation.error