""" echo/core/orchestrator.py ------------------------- The conductor. Wires Curator + Screenwriter + Verifier + tools into the loop that grows the tree of alternate lives. Branch-growth loop (one user choice -> one new node): 1. Curator reads parent + branch history (+ research grounding) -> child 2. Verifier audits child vs branch -> regenerate up to N times if needed 3. Voice tool speaks the child's line 4. Screenwriter plans the child's next two forks 5. node added to the tree The orchestrator is UI-agnostic: the Gradio app calls seed() once, then choose_fork() each time the user clicks a branch. Everything it returns is plain data the front-end renders. """ from __future__ import annotations from dataclasses import dataclass from typing import Optional from .world_state import WorldState, root_state from .tree import LifeTree from ..agents.curator import Curator from ..agents.screenwriter import Screenwriter from ..agents.verifier import Verifier from ..tools.research import ResearchTool from ..tools.voice import VoiceTool from ..llm.client import LLMClient @dataclass class GrowthStats: """Telemetry — powers the Tiny Titan experiment (regeneration rate).""" nodes_grown: int = 0 regenerations: int = 0 @property def regen_rate(self) -> float: denom = self.nodes_grown + self.regenerations return self.regenerations / denom if denom else 0.0 class Orchestrator: def __init__(self, llm: LLMClient, research: ResearchTool, voice: VoiceTool, out_dir: str = "echo_out", max_regen: int = 2, base_year: int = 2025): self.curator = Curator(llm) self.screenwriter = Screenwriter(llm) self.verifier = Verifier(llm) self.research = research self.voice = voice self.out_dir = out_dir self.max_regen = max_regen self.base_year = base_year self.tree: Optional[LifeTree] = None self.stats = GrowthStats() # ------------------------------------------------------------------ seed def seed(self, seed_choice: str, base_age: int = 30) -> WorldState: """Plant the tree: create the root life and its first two forks.""" self.tree = LifeTree(seed_choice=seed_choice) root = root_state(seed_choice, base_age) # The root itself is curated from the seed (depth 0, 0 years elapsed). grounding = self.research.ground(root.facts.location, self.base_year) curated = self.curator.grow( parent=root, branch_narrative=f"Origin: {seed_choice}", chosen_fork=seed_choice, years=0, grounding=grounding, ) curated.parent_id = None curated.depth = 0 curated.divergence = seed_choice self._voice(curated) curated.pending_forks = self.screenwriter.plan( curated, self.tree.branch_narrative(curated.node_id) if curated.node_id in self.tree.nodes else f"Origin: {seed_choice}" ) self.tree.add(curated) # plan forks again now that it's in the tree (branch narrative valid) curated.pending_forks = self.screenwriter.plan( curated, self.tree.branch_narrative(curated.node_id)) self.stats.nodes_grown += 1 return curated # ----------------------------------------------------------- grow branch def choose_fork(self, parent_id: str, fork_index: int, years: int = 5) -> WorldState: """User picks one of a node's pending forks -> grow the child node.""" assert self.tree is not None, "call seed() first" parent = self.tree.nodes[parent_id] fork = parent.pending_forks[fork_index] narrative = self.tree.branch_narrative(parent_id) child = self._grow_verified(parent, narrative, fork, years) self._voice(child) child.pending_forks = self.screenwriter.plan( child, self.tree.branch_narrative(parent_id) + f"\n-> {fork}") self.tree.add(child) self.stats.nodes_grown += 1 return child # ----------------------------------------------------- internal helpers def _grow_verified(self, parent: WorldState, narrative: str, fork: str, years: int) -> WorldState: """Curator + Verifier loop with bounded regeneration.""" location_year = self.base_year + (parent.depth + 1) * years grounding = self.research.ground(parent.facts.location, location_year) child = self.curator.grow(parent, narrative, fork, years, grounding) # Verify EVERY attempt, including the last regeneration. (do-while) # Breaking out with an unverified child is how a flagged node used to # slip into the tree silently. for attempt in range(self.max_regen + 1): verdict = self.verifier.check(child, parent, narrative) if verdict.consistent or attempt == self.max_regen: break self.stats.regenerations += 1 # regenerate with the contradiction noted in the grounding hint child = self.curator.grow( parent, narrative, fork, years, grounding=(grounding or "") + f"\nFIX: previous attempt was inconsistent ({verdict.reason}).") return child def _voice(self, state: WorldState) -> None: if state.voice_line: state.voice_audio_path = self.voice.speak(state, self.out_dir) # ----------------------------------------------------------------- views def graph(self) -> dict: return self.tree.to_graph() if self.tree else {"nodes": [], "edges": []} def final_map_summary(self) -> str: """The closing artifact: a gentle celebration of the lives explored.""" if not self.tree: return "" n = self.tree.size flourishing = sum(1 for x in self.tree.nodes.values() if x.tone.is_flourishing) return (f"You explored {n} lives that all lived inside one choice. " f"{flourishing} of them were flourishing. " f"None of them are more real than the one you're in.")