from dotenv import load_dotenv
load_dotenv()
from tensorlake.sandbox import SandboxClient
from pydantic import BaseModel
from typing import List, Optional
from openai import OpenAI
from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from rich.rule import Rule
from rich import box
import json
from concurrent.futures import ThreadPoolExecutor
import time
console = Console()
# ─── Data Models ─────────────────────────────────────────────────────────────
class PlayerState(BaseModel):
player_name: str
hp: int = 20
max_hp: int = 20
inventory: List[str] = ["torch", "dagger"]
story_history: List[str] = []
current_choice: Optional[str] = None
turn: int = 0
class SceneDraft(BaseModel):
branch_id: int
branch_label: str
narrative: str
consequences: str
image_prompt: str
ascii_art: str
class StoryBeat(BaseModel):
scene_narrative: str
choices: List[str]
image_prompt: str
ascii_art: str
updated_state: PlayerState
# ─── Agent 1: Scene Writer (runs in parallel per branch) ─────────────────────
def scene_agent(args: dict) -> SceneDraft:
"""
Each scene_agent drafts ONE possible branch outcome in an isolated sandbox.
Runs in parallel — one sandbox per branch.
"""
branch_id = args["branch_id"]
branch_label = args["branch_label"]
player_state = PlayerState(**args["player_state"])
setting = args["setting"]
print(f"⚔️ Scene Agent [{branch_label}]: Drafting branch in sandbox...")
client = OpenAI()
prompt = f"""
You are a D&D scene writer. The player chose: "{branch_label}".
Setting: {setting}
Player: {player_state.player_name}, HP: {player_state.hp}/{player_state.max_hp}
Inventory: {player_state.inventory}
Story so far: {' | '.join(player_state.story_history[-3:]) or 'Adventure begins.'}
Write a Python script that:
1. Uses 'random' to simulate a D20 dice roll
2. Determines success/failure of the action "{branch_label}" based on the roll (>=10 is success)
3. Prints a single valid JSON (no markdown, no extra text) with these exact keys:
- branch_id: {branch_id}
- branch_label: "{branch_label}"
- narrative: vivid 3-sentence scene description of the outcome
- consequences: one of "-N HP", "+item_name", "no change", or "unlocked secret"
- image_prompt: a DALL-E prompt for this scene in dark fantasy style
- ascii_art: a 10-15 line ASCII art illustration using / \\ | _ . * # @ ~ ^
that depicts the scene visually. Must be a single string with \\n for newlines.
Make it evocative of the environment — dungeon, dragon, forest, castle, monster, etc.
IMPORTANT: The script must print ONLY valid JSON to stdout. No markdown, no code fences.
"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
generated_code = (
response.choices[0].message.content
.replace("```python", "")
.replace("```", "")
.strip()
)
print(f"⚔️ Scene Agent [{branch_label}]: Executing dice logic in Sandbox...")
sb_client = SandboxClient()
with sb_client.create_and_connect() as sandbox:
execution = sandbox.run("python3", ["-c", generated_code])
output = execution.stdout.strip()
print(f"⚔️ Scene Agent [{branch_label}]: Result -> {output[:80]}...")
data = json.loads(output)
return SceneDraft(**data)
# ─── Agent 2: Dungeon Master (aggregator + narrator) ─────────────────────────
def dungeon_master(args: dict) -> StoryBeat:
"""
The DM receives all parallel branch drafts, picks the player's chosen one,
narrates the next scene with 3 new choices, and updates state.
"""
drafts = [SceneDraft(**d) for d in args["drafts"]]
player_state = PlayerState(**args["player_state"])
chosen_label = player_state.current_choice
print(f"🎲 Dungeon Master: Received {len(drafts)} branch drafts. Chosen: '{chosen_label}'")
chosen = next((d for d in drafts if d.branch_label == chosen_label), drafts[0])
client = OpenAI()
prompt = f"""
You are an epic Dungeon Master continuing a D&D adventure.
The player chose: "{chosen.branch_label}"
What happened: {chosen.narrative}
Consequences: {chosen.consequences}
Player state: HP={player_state.hp}/{player_state.max_hp}, Inventory={player_state.inventory}
Turn number: {player_state.turn}
Now write the next story beat. Respond ONLY as raw JSON (no markdown, no code fences) with:
- scene_narrative: 2 vivid paragraphs in second person ("You...") describing what unfolds
- choices: list of exactly 3 short action choices for the player (action verbs, max 5 words each)
- image_prompt: a DALL-E prompt for the scene illustration in dark fantasy oil painting style
"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
raw = (
response.choices[0].message.content
.replace("```json", "")
.replace("```", "")
.strip()
)
data = json.loads(raw)
# Apply consequences to player state
updated = player_state.model_copy(deep=True)
cons = chosen.consequences.lower()
if "hp" in cons and "-" in cons:
try:
dmg = int(''.join(filter(str.isdigit, cons.split("hp")[0])))
updated.hp = max(0, updated.hp - dmg)
except ValueError:
pass
elif "hp" in cons and "+" in cons:
try:
heal = int(''.join(filter(str.isdigit, cons.split("hp")[0])))
updated.hp = min(updated.max_hp, updated.hp + heal)
except ValueError:
pass
if "+" in cons and "hp" not in cons:
item = chosen.consequences.replace("+", "").strip()
if item and item not in updated.inventory:
updated.inventory.append(item)
updated.story_history.append(chosen.narrative[:100])
updated.turn += 1
return StoryBeat(
scene_narrative=data["scene_narrative"],
choices=data["choices"],
image_prompt=data["image_prompt"],
ascii_art=chosen.ascii_art,
updated_state=updated,
)
# ─── Application: One Full RPG Turn ──────────────────────────────────────────
def rpg_adventure(player_name: str, choice: str, state_json: str, setting: str) -> str:
"""
One full turn:
1. Fan out 3 parallel scene agents (one per branch)
2. DM aggregates and narrates the chosen branch
3. Returns next StoryBeat as JSON
"""
print(f"\n🧙 RPG Turn: Player='{player_name}', Choice='{choice}'")
state = json.loads(state_json)
state["current_choice"] = choice
branches = [
{"branch_id": 0, "branch_label": "Fight"},
{"branch_id": 1, "branch_label": "Flee"},
{"branch_id": 2, "branch_label": "Negotiate"},
]
scene_args = [
{**b, "player_state": state, "setting": setting}
for b in branches
]
# Threads are needed because the local SandboxClient blocks while waiting for output.
# The sandboxes THEMSELVES run in parallel on the server, but threads allow
# our script to wait for multiple results simultaneously.
with ThreadPoolExecutor(max_workers=len(branches)) as executor:
drafts = list(executor.map(scene_agent, scene_args))
beat = dungeon_master({
"drafts": [d.model_dump() for d in drafts],
"player_state": state,
})
return beat.model_dump_json(indent=2)
# ─── UI Helpers ──────────────────────────────────────────────────────────────
TITLE_SCREEN = r"""
____ ____ _ __ ___ __ ______________ _________
/ __ \/ __ \/ | / / / _ | / / / / __/ ___/ _ \/ ___/ __/
/ /_/ / / / / |/ / / __ |/ /_/ / _// (_ / // / /__/ _/
\____/_/ /_/_/|_/ /_/ |_|\____/___/\___/____/\___/___/
🐉 A N A I - P O W E R E D A D V E N T U R E 🐉
"""
def print_title():
console.print()
console.print(Text(TITLE_SCREEN, style="bold red"))
console.print(Rule(style="red"))
console.print()
def print_scene(beat: StoryBeat):
console.print()
console.print(Rule("⚔ NEW SCENE", style="yellow"))
# ASCII art panel
console.print(
Panel(
Text(beat.ascii_art, style="bold green", justify="center"),
border_style="dim green",
padding=(1, 4),
)
)
# Narrative panel
console.print(
Panel(
beat.scene_narrative,
title="[bold cyan]📖 What Unfolds[/bold cyan]",
border_style="cyan",
padding=(1, 2),
)
)
def print_stats(state: PlayerState):
hp_color = "bold green" if state.hp > 10 else "bold yellow" if state.hp > 5 else "bold red"
hp_bar = "█" * state.hp + "░" * (state.max_hp - state.hp)
inv_str = ", ".join(state.inventory) if state.inventory else "nothing"
console.print(
Panel(
f"[{hp_color}]❤ HP: {state.hp}/{state.max_hp} [{hp_bar}][/{hp_color}]\n"
f"[bold white]🎒 Inventory:[/bold white] [dim]{inv_str}[/dim]\n"
f"[bold white]📜 Turn:[/bold white] [dim]{state.turn}[/dim]",
title=f"[bold magenta]🧙 {state.player_name}[/bold magenta]",
border_style="magenta",
box=box.SIMPLE,
)
)
def print_choices(choices: List[str]):
console.print()
console.print(Rule("🎮 YOUR MOVE", style="bold yellow"))
for i, c in enumerate(choices, 1):
console.print(f" [bold yellow]{i}.[/bold yellow] [white]{c}[/white]")
console.print(f" [dim]q. Quit adventure[/dim]")
console.print()
def get_player_choice(choices: List[str]) -> Optional[str]:
while True:
console.print("[bold]Enter 1, 2, 3 or q:[/bold] ", end="")
raw = input().strip().lower()
if raw == "q":
return None
if raw in ("1", "2", "3"):
idx = int(raw) - 1
if idx < len(choices):
return choices[idx]
console.print(" [bold red]⚠ Invalid input. Try 1, 2, 3 or q.[/bold red]")
# ─── Entry Point ─────────────────────────────────────────────────────────────
if __name__ == "__main__":
print_title()
# Hero name
console.print("[bold]Enter your hero's name[/bold] (or press Enter for 'Aldric the Bold'): ", end="")
player_name = input().strip() or "Aldric the Bold"
console.print(f"\n[bold green]Welcome, {player_name}! Your legend begins...[/bold green]\n")
time.sleep(1)
# Initial state & setting
state = PlayerState(player_name=player_name)
setting = (
"A crumbling dungeon entrance lit by sickly green torchlight. "
"Ancient runes glow on the walls. Something massive growls in the darkness ahead. "
"The air smells of sulfur and old bones."
)
current_choice = "Explore"
# ── Game Loop ──
while True:
console.print(f"\n[dim]⏳ Generating scene for: '[italic]{current_choice}[/italic]'...[/dim]")
try:
# Directly call the function instead of using run_local_application
result_json = rpg_adventure(
player_name=player_name,
choice=current_choice,
state_json=state.model_dump_json(),
setting=setting,
)
beat = StoryBeat.model_validate_json(result_json)
except Exception as e:
console.print(f"\n[bold red]❌ Error generating scene: {e}[/bold red]")
console.print("[dim]Retrying...[/dim]")
continue
# Update state
state = beat.updated_state
# Render scene
print_scene(beat)
print_stats(state)
# Death check
if state.hp <= 0:
console.print(
Panel(
"[bold red]💀 You have fallen in battle.\n\nYour legend ends here... for now.[/bold red]",
border_style="red",
padding=(1, 2),
)
)
break
# Choices
print_choices(beat.choices)
chosen = get_player_choice(beat.choices)
if chosen is None:
console.print(
"\n[bold yellow]🏰 You sheathe your sword and walk away into the mist.\n"
"Farewell, adventurer. Your story is unfinished.[/bold yellow]\n"
)
break
# Roll forward
current_choice = chosen
setting = beat.scene_narrative[-300:] # tail of scene becomes new setting context
