import copy import html import json import os import re import time import traceback from pathlib import Path import gradio as gr PROJECT_ROOT = Path(__file__).parent ASSETS_DIR = PROJECT_ROOT / "assets" ASSET_CATALOG_PATH = ASSETS_DIR / "asset_catalog.json" RENDERER_VERSION = "0.5" WORLD_SCHEMA_VERSION = "pocketworld-world-v0.5" ASSET_SCHEMA_VERSION = "pocketworld-assets-v0.1" TEXT_WORLD_MODEL_ID = "openbmb/MiniCPM5-1B" AI_DISABLE_ENV = "POCKETWORLD_DISABLE_AI" AI_FORCE_FAIL_ENV = "POCKETWORLD_FORCE_AI_FAIL" AI_SELF_CHECK_ENV = "POCKETWORLD_PHASE4A_SELF_CHECK" AI_MAX_GENERATION_SECONDS_ENV = "POCKETWORLD_MAX_GENERATION_SECONDS" AI_PROMPT_MODE_ENV = "POCKETWORLD_PROMPT_MODE" AI_USE_JSON_PREFILL_ENV = "POCKETWORLD_USE_JSON_PREFILL" AI_DEBUG_MODEL_OUTPUT_ENV = "POCKETWORLD_DEBUG_MODEL_OUTPUT" AI_MODE_ENV = "POCKETWORLD_AI_MODE" DEFAULT_AI_MODE = "design_patch" THEME_OPTIONS = ["Auto", "Light", "Dark"] DEFAULT_THEME = "Auto" WORLD_THEMES = ["cozy_fantasy", "sci_fi_station", "haunted_mystery", "tiny_city"] DEFAULT_WORLD_THEME = "cozy_fantasy" AI_SOURCE_SCENES = ["Chaotic Desk", "Cozy Kitchen", "Lab Bench", "City Corner"] AI_GENRES = ["Cozy Fantasy", "Sci-Fi", "Haunted Mystery", "Tiny City"] AI_TONES = ["Whimsical", "Epic", "Eerie", "Funny"] AI_GAME_TEMPLATES = ["Quest Gate", "Mystery Trail", "Rescue Route"] AI_WORLD_SIZES = ["Medium", "Large"] _TEXT_MODEL_CACHE = {"tokenizer": None, "model": None, "torch": None} TILE_LEGEND = { "W": "wall / blocked", ".": "floor / walkable", "G": "locked goal or exit", } EMBEDDED_ASSET_CATALOG = { "schema_version": ASSET_SCHEMA_VERSION, "source": { "name": "Kenney Tiny Dungeon", "url": "https://kenney.nl/assets/tiny-dungeon", "license": "Creative Commons Zero (CC0)", "license_url": "https://creativecommons.org/publicdomain/zero/1.0/", "credit": "Kenney", }, "tile_size": 16, "display_tile_size": 44, "themes": { "cozy_fantasy": { "tile_palette": {"W": "wall_wood", ".": "floor_wood", "G": "gate"}, "player_sprite_key": "player", "npc_sprite_keys": ["npc_wizard", "npc_merchant", "npc_citizen"], "item_sprite_keys": ["key", "gem", "potion", "scroll"], "landmark_asset_keys": ["gate", "well", "tower", "bridge"], }, "sci_fi_station": { "tile_palette": {"W": "wall_metal", ".": "floor_metal", "G": "portal"}, "player_sprite_key": "player", "npc_sprite_keys": ["npc_robot", "npc_scientist"], "item_sprite_keys": ["battery", "gear", "tool"], "landmark_asset_keys": ["portal", "computer", "door"], }, "haunted_mystery": { "tile_palette": {"W": "wall_stone", ".": "floor_stone", "G": "door"}, "player_sprite_key": "player", "npc_sprite_keys": ["npc_detective", "npc_librarian"], "item_sprite_keys": ["book", "note", "key"], "landmark_asset_keys": ["door", "shelf", "lamp"], }, "tiny_city": { "tile_palette": {"W": "wall_brick", ".": "floor_city", "G": "sign"}, "player_sprite_key": "player", "npc_sprite_keys": ["npc_citizen", "npc_merchant", "npc_robot"], "item_sprite_keys": ["coin", "tool", "battery"], "landmark_asset_keys": ["sign", "bridge", "computer"], }, }, "assets": { "tiles": { "floor_grass": {"path": "assets/tiles/floor_grass.png", "license": "CC0"}, "floor_stone": {"path": "assets/tiles/floor_stone.png", "license": "CC0"}, "floor_wood": {"path": "assets/tiles/floor_wood.png", "license": "CC0"}, "floor_metal": {"path": "assets/tiles/floor_metal.png", "license": "CC0"}, "floor_city": {"path": "assets/tiles/floor_city.png", "license": "CC0"}, "wall_stone": {"path": "assets/tiles/wall_stone.png", "license": "CC0"}, "wall_wood": {"path": "assets/tiles/wall_wood.png", "license": "CC0"}, "wall_metal": {"path": "assets/tiles/wall_metal.png", "license": "CC0"}, "wall_brick": {"path": "assets/tiles/wall_brick.png", "license": "CC0"}, "path_dirt": {"path": "assets/tiles/path_dirt.png", "license": "CC0"}, "path_cable": {"path": "assets/tiles/path_cable.png", "license": "CC0"}, "water": {"path": "assets/tiles/water.png", "license": "CC0"}, }, "chars": { "player": {"path": "assets/chars/player.png", "license": "CC0"}, "npc_wizard": {"path": "assets/chars/npc_wizard.png", "license": "CC0"}, "npc_robot": {"path": "assets/chars/npc_robot.png", "license": "CC0"}, "npc_merchant": {"path": "assets/chars/npc_merchant.png", "license": "CC0"}, "npc_librarian": {"path": "assets/chars/npc_librarian.png", "license": "CC0"}, "npc_detective": {"path": "assets/chars/npc_detective.png", "license": "CC0"}, "npc_scientist": {"path": "assets/chars/npc_scientist.png", "license": "CC0"}, "npc_citizen": {"path": "assets/chars/npc_citizen.png", "license": "CC0"}, }, "items": { "key": {"path": "assets/items/key.png", "license": "CC0"}, "book": {"path": "assets/items/book.png", "license": "CC0"}, "gem": {"path": "assets/items/gem.png", "license": "CC0"}, "potion": {"path": "assets/items/potion.png", "license": "CC0"}, "coin": {"path": "assets/items/coin.png", "license": "CC0"}, "scroll": {"path": "assets/items/scroll.png", "license": "CC0"}, "battery": {"path": "assets/items/battery.png", "license": "CC0"}, "gear": {"path": "assets/items/gear.png", "license": "CC0"}, "tool": {"path": "assets/items/tool.png", "license": "CC0"}, "note": {"path": "assets/items/note.png", "license": "CC0"}, }, "landmarks": { "gate": {"path": "assets/landmarks/gate.png", "license": "CC0"}, "well": {"path": "assets/landmarks/well.png", "license": "CC0"}, "tower": {"path": "assets/landmarks/tower.png", "license": "CC0"}, "portal": {"path": "assets/landmarks/portal.png", "license": "CC0"}, "door": {"path": "assets/landmarks/door.png", "license": "CC0"}, "computer": {"path": "assets/landmarks/computer.png", "license": "CC0"}, "shelf": {"path": "assets/landmarks/shelf.png", "license": "CC0"}, "sign": {"path": "assets/landmarks/sign.png", "license": "CC0"}, "lamp": {"path": "assets/landmarks/lamp.png", "license": "CC0"}, "bridge": {"path": "assets/landmarks/bridge.png", "license": "CC0"}, }, }, } DETECTED_OBJECT_SCHEMA = { "image_id": "optional str", "objects": [ { "id": "optional str", "label": "str, for example 'coffee mug'", "confidence": "optional float 0..1", "bbox": "optional [x, y, width, height] in source image pixels", "attributes": "optional dict from a vision model", } ], } WORLD_SCHEMA = { "schema_version": WORLD_SCHEMA_VERSION, "title": "str", "genre": "str", "intro": "optional str shown before gameplay starts", "theme": WORLD_THEMES, "source": { "kind": "sample | image_objects | text_prompt | model_generated", "objects": "optional normalized object list using DETECTED_OBJECT_SCHEMA", }, "tiles": { "legend": TILE_LEGEND, "rows": "list[str] with equal width; use W, ., and G", }, "tile_palette": {"W": "asset_key", ".": "asset_key", "G": "asset_key"}, "player_start": "[x, y]", "player_sprite_key": "asset key from chars", "regions": [ {"id": "str", "name": "str", "source_object": "str", "description": "str"} ], "grounding": [ { "source_object": "str", "world_object": "str", "role": "str", "asset_key": "asset key from the fixed catalog", } ], "npcs": [ { "id": "str", "name": "str", "x": "int", "y": "int", "sprite_key": "asset key from chars", "role": "guide | blocker | lorekeeper | trickster | merchant", "dialogue": "str", } ], "items": [ { "id": "str", "name": "str", "x": "int", "y": "int", "sprite_key": "asset key from items", "description": "str", } ], "landmarks": [ { "id": "str", "name": "str", "x": "int", "y": "int", "sprite_key": "asset key from landmarks", "source_object": "str", "description": "str", } ], "quest": { "goal": "str", "goal_id": "optional id for the locked gate or final objective", "required_item": "item id", "success_ending": "str", }, "quest_steps": [ {"id": "talk_guide", "type": "talk", "target": "npc id", "text": "Talk to the guide."}, {"id": "inspect_archive", "type": "inspect", "target": "landmark id", "text": "Inspect the landmark."}, {"id": "collect_key", "type": "collect", "target": "item id", "text": "Find the key item."}, {"id": "unlock_gate", "type": "unlock", "target": "goal id or goal", "text": "Unlock the gate."}, ], } MODEL_BOUNDARY_CONTRACT = { "renderer_version": RENDERER_VERSION, "input_from_future_vision_model": DETECTED_OBJECT_SCHEMA, "input_from_future_world_model": WORLD_SCHEMA, "model_rule": "Choose only asset keys from asset_catalog.json. Never invent filenames.", "renderer_guarantees": [ "No Python callbacks during gameplay.", "World JSON is normalized before rendering.", "Unknown asset keys become warnings and theme fallback keys.", "Missing asset PNG files fall back to simple Phaser shapes.", "Invalid map and quest playability still raise clear Gradio errors.", ], } MOCK_SCENE_PARSES = { "Chaotic Desk": { "source_scene": "Chaotic Desk", "scene_type": "desk", "mood": "busy, academic, slightly chaotic", "objects": [ {"name": "coffee mug", "location": "front left", "role_hint": "container, harbor, or well"}, {"name": "notebook", "location": "right side", "role_hint": "archive, clue, or tower"}, {"name": "laptop", "location": "center", "role_hint": "gate, terminal, or final portal"}, {"name": "charger cable", "location": "bottom edge", "role_hint": "road, bridge, or serpent path"}, {"name": "sticky note", "location": "near notebook", "role_hint": "key, message, or quest clue"}, {"name": "desk lamp", "location": "back corner", "role_hint": "landmark, beacon, or moonwell"}, ], }, "Cozy Kitchen": { "source_scene": "Cozy Kitchen", "scene_type": "kitchen counter", "mood": "warm, practical, homelike", "objects": [ {"name": "tea kettle", "location": "back left", "role_hint": "tower, well, or guide house"}, {"name": "cutting board", "location": "center", "role_hint": "plaza, bridge, or village square"}, {"name": "spice jar", "location": "right shelf", "role_hint": "magic item, clue, or potion"}, {"name": "recipe card", "location": "front counter", "role_hint": "archive, map, or key"}, {"name": "oven door", "location": "back wall", "role_hint": "locked gate or final door"}, {"name": "wooden spoon", "location": "front right", "role_hint": "tool, wand, or optional charm"}, ], }, "Lab Bench": { "source_scene": "Lab Bench", "scene_type": "science workbench", "mood": "technical, bright, experimental", "objects": [ {"name": "microscope", "location": "center left", "role_hint": "observatory, tower, or lore landmark"}, {"name": "test tube rack", "location": "center", "role_hint": "village, battery bank, or puzzle station"}, {"name": "gloves", "location": "front edge", "role_hint": "tool or protective key"}, {"name": "tablet", "location": "right side", "role_hint": "terminal, gate, or portal"}, {"name": "cable coil", "location": "bottom left", "role_hint": "road, bridge, or loop"}, {"name": "sample vial", "location": "near rack", "role_hint": "required item, potion, or power cell"}, ], }, "City Corner": { "source_scene": "City Corner", "scene_type": "street corner", "mood": "compact, urban, lively", "objects": [ {"name": "street sign", "location": "corner pole", "role_hint": "final gate, landmark, or guide post"}, {"name": "bench", "location": "left sidewalk", "role_hint": "village, rest area, or NPC spot"}, {"name": "bike lock", "location": "front rail", "role_hint": "required key item or lock"}, {"name": "newspaper box", "location": "right side", "role_hint": "archive, clue box, or merchant stall"}, {"name": "crosswalk", "location": "bottom edge", "role_hint": "path, bridge, or road"}, {"name": "apartment door", "location": "back wall", "role_hint": "goal door or final threshold"}, ], }, } AI_GENRE_CONFIG = { "Cozy Fantasy": {"theme": "cozy_fantasy", "genre": "cozy fantasy micro-quest"}, "Sci-Fi": {"theme": "sci_fi_station", "genre": "sci-fi station micro-quest"}, "Haunted Mystery": {"theme": "haunted_mystery", "genre": "haunted mystery micro-quest"}, "Tiny City": {"theme": "tiny_city", "genre": "tiny city micro-quest"}, } AI_TEMPLATE_RULES = { "Quest Gate": "Classic gate quest: talk to a guide, inspect one landmark, collect the required item, then unlock the final gate.", "Mystery Trail": "Mystery trail: talk to a clue-giver, inspect a clue landmark, collect the discovered evidence item, then unlock the final threshold.", "Rescue Route": "Rescue route: talk to a helper, inspect the place where someone is stuck, collect a rescue tool, then open the final exit.", } DESIGN_PATCH_SCHEMA = { "title": "string", "intro": "string", "world_tone": "string", "transformations": [ { "source_object": "coffee mug", "world_object": "Moonwell Harbor", "role": "landmark", "why": "Because it looks like a container of liquid.", } ], "npc_names": ["string", "string"], "landmark_names": ["string", "string"], "item_names": ["string", "string"], "quest_lines": [ "Talk to the guide.", "Inspect the archive.", "Collect the key.", "Unlock the gate.", ], "dialogue_lines": ["string", "string", "string"], "ending": "string", } MEDIUM_TEMPLATE_TILES = [ "WWWWWWWWWWWWWWWWWWWW", "W..................W", "W..WW.....W........W", "W...W.....W..WW....W", "W..................W", "W..WWW.............W", "W......W...........W", "W......W..WWW......W", "W..................W", "W....WWWW..........W", "W..................W", "W.............WW...W", "W.................GW", "WWWWWWWWWWWWWWWWWWWW", ] LARGE_TEMPLATE_TILES = [ "WWWWWWWWWWWWWWWWWWWWWWWW", "W......................W", "W......W...............W", "W......W...............W", "W......W...............W", "W......W...............W", "W......................W", "W......................W", "W......................W", "W..............W.......W", "W..............W.......W", "W..............W.......W", "W..............W.......W", "W.....................GW", "W......................W", "WWWWWWWWWWWWWWWWWWWWWWWW", ] WORLD_BUILDER_TEMPLATES = { "medium_quest_gate": { "tiles": MEDIUM_TEMPLATE_TILES, "player_start": [2, 11], "npc_slots": [[3, 10], [5, 2]], "landmark_slots": [[10, 8], [15, 3]], "item_slots": [[16, 10], [4, 4]], "gate_slot": [18, 12], "regions": [ ["arrival_path", "Arrival Path", "A safe route into the pocket world."], ["clue_corner", "Clue Corner", "A small area where the guide explains the first task."], ["gate_reach", "Gate Reach", "The final path toward the locked gate."], ], }, "large_quest_gate": { "tiles": LARGE_TEMPLATE_TILES, "player_start": [2, 13], "npc_slots": [[3, 12], [5, 2]], "landmark_slots": [[12, 8], [19, 4]], "item_slots": [[20, 3], [4, 4]], "gate_slot": [22, 13], "regions": [ ["arrival_path", "Arrival Path", "A longer route into the pocket world."], ["guide_crossing", "Guide Crossing", "A social pocket where the first hint appears."], ["gate_reach", "Gate Reach", "A final stretch near the locked gate."], ], }, "medium_mystery_trail": { "tiles": MEDIUM_TEMPLATE_TILES, "player_start": [2, 11], "npc_slots": [[3, 10], [14, 2]], "landmark_slots": [[8, 4], [15, 8]], "item_slots": [[16, 10], [4, 4]], "gate_slot": [18, 12], "regions": [ ["trailhead", "Trailhead", "The first clue along the mystery trail."], ["archive_bend", "Archive Bend", "A landmark bend where evidence collects."], ["index_gate", "Index Gate", "The locked ending of the trail."], ], }, "large_mystery_trail": { "tiles": LARGE_TEMPLATE_TILES, "player_start": [2, 13], "npc_slots": [[3, 12], [18, 2]], "landmark_slots": [[12, 8], [19, 4]], "item_slots": [[20, 3], [4, 4]], "gate_slot": [22, 13], "regions": [ ["trailhead", "Trailhead", "A broad entry into the mystery trail."], ["clue_archive", "Clue Archive", "A landmark area holding the key clue."], ["index_gate", "Index Gate", "The locked ending of the trail."], ], }, "medium_rescue_route": { "tiles": MEDIUM_TEMPLATE_TILES, "player_start": [2, 11], "npc_slots": [[3, 10], [5, 2]], "landmark_slots": [[10, 8], [15, 3]], "item_slots": [[16, 10], [4, 4]], "gate_slot": [18, 12], "regions": [ ["rescue_start", "Rescue Start", "The beginning of a tiny rescue route."], ["signal_post", "Signal Post", "A landmark that explains who needs help."], ["safe_exit", "Safe Exit", "The locked route to safety."], ], }, "large_rescue_route": { "tiles": LARGE_TEMPLATE_TILES, "player_start": [2, 13], "npc_slots": [[3, 12], [5, 2]], "landmark_slots": [[12, 8], [19, 4]], "item_slots": [[20, 3], [4, 4]], "gate_slot": [22, 13], "regions": [ ["rescue_start", "Rescue Start", "The beginning of a wider rescue route."], ["signal_post", "Signal Post", "A landmark that explains who needs help."], ["safe_exit", "Safe Exit", "The locked route to safety."], ], }, } SAMPLE_WORLDS = [ { "title": "Moonwell Harbor", "genre": "cozy desk fantasy", "intro": "A tiny harbor has formed from a chaotic desk: mug piers, cable roads, notebook towers, and one stubborn moonlit gate.", "theme": "cozy_fantasy", "tile_palette": {"W": "wall_wood", ".": "floor_wood", "G": "gate"}, "player_sprite_key": "player", "tiles": [ "WWWWWWWWWWWWWWWWWWWWWWWW", "W......W...............W", "W..WW..W.......W.......W", "W......W...WWW.W.......W", "W......W...............W", "W......W...............W", "W..WW.WWWWWW...W.......W", "W......W.......W....WW.W", "W......W.......W.......W", "W..WWW.W.......W.......W", "W......W....WWWWW.WWWW.W", "W......W.......W.......W", "W......................W", "W.....................GW", "W..............W.......W", "WWWWWWWWWWWWWWWWWWWWWWWW", ], "player_start": [2, 13], "regions": [ { "id": "starting_cove", "name": "Starting Cove", "source_object": "desk corner", "description": "A sheltered cove where the desk clutter opens into a tiny path.", }, { "id": "cupstone_village", "name": "Cupstone Village", "source_object": "coffee mug", "description": "A small guide village tucked beside the curved wall of a coffee mug.", }, { "id": "archive_lane", "name": "Archive Lane", "source_object": "notebook", "description": "A quiet route past page-like shelves and note-stacked walls.", }, { "id": "moonwell_gate", "name": "Moonwell Gate", "source_object": "desk lamp", "description": "A violet-lit final gate beyond the cable road.", }, ], "grounding": [ {"source_object": "coffee mug", "world_object": "Cupstone Village", "role": "starting village", "asset_key": "well"}, {"source_object": "notebook", "world_object": "Archive of Lost Plans", "role": "inspectable landmark", "asset_key": "tower"}, {"source_object": "desk lamp", "world_object": "Moonwell Lamp", "role": "inspectable landmark", "asset_key": "lamp"}, {"source_object": "sticky note", "world_object": "Paper Key", "role": "required item", "asset_key": "key"}, {"source_object": "paperclip", "world_object": "Silver Clip", "role": "optional item", "asset_key": "gem"}, {"source_object": "laptop glow", "world_object": "Moonwell Gate", "role": "final gate", "asset_key": "gate"}, ], "npcs": [ { "id": "guide", "name": "Tidekeeper Nima", "x": 4, "y": 12, "sprite_key": "npc_wizard", "role": "guide", "dialogue": "Welcome to Moonwell Harbor. The gate will not listen until the Archive remembers your name.", }, { "id": "merchant", "name": "Clip Merchant Orro", "x": 3, "y": 3, "sprite_key": "npc_merchant", "role": "merchant", "dialogue": "The Paper Key is east of the old archive, but a Silver Clip never hurts morale.", }, ], "items": [ { "id": "silver_clip", "name": "Silver Clip", "x": 5, "y": 5, "sprite_key": "gem", "description": "An optional bright paperclip charm. It is not the key, but it feels lucky.", }, { "id": "paper_key", "name": "Paper Key", "x": 20, "y": 3, "sprite_key": "key", "description": "A folded sticky-note key stamped with a tiny moon.", }, ], "landmarks": [ { "id": "archive", "name": "Archive of Lost Plans", "x": 10, "y": 8, "sprite_key": "tower", "source_object": "notebook", "description": "A quiet archive made from the notebook in the source image. Its shelves whisper the route to the Paper Key.", }, { "id": "moonwell_lamp", "name": "Moonwell Lamp", "x": 18, "y": 5, "sprite_key": "lamp", "source_object": "desk lamp", "description": "A little lamp-landmark throwing blue light over the road to the final gate.", }, ], "quest": { "goal": "Follow the desk-world trail, recover the Paper Key, and open the Moonwell Gate.", "goal_id": "blue_gate", "required_item": "paper_key", "success_ending": "The Moonwell Gate blooms open, and the tiny harbor finds its tide.", }, "quest_steps": [ {"id": "talk_guide", "type": "talk", "target": "guide", "text": "Talk to Tidekeeper Nima in Cupstone Village."}, {"id": "inspect_archive", "type": "inspect", "target": "archive", "text": "Inspect the Archive of Lost Plans."}, {"id": "collect_key", "type": "collect", "target": "paper_key", "text": "Find the Paper Key beyond Archive Lane."}, {"id": "unlock_gate", "type": "unlock", "target": "blue_gate", "text": "Unlock the Moonwell Gate."}, ], }, { "title": "Blue Screen Station", "genre": "sci-fi station errand", "theme": "sci_fi_station", "tile_palette": {"W": "wall_metal", ".": "floor_metal", "G": "portal"}, "player_sprite_key": "player", "tiles": [ "WWWWWWWWWWWW", "W..........W", "W.WWWW..W..W", "W.W.....W..W", "W.W..W.....W", "W....W..G..W", "W..........W", "WWWWWWWWWWWW", ], "player_start": [1, 1], "regions": [ {"id": "cable_road", "name": "Black Cable Road", "source_object": "charger cable", "description": "A cable-like corridor across a tiny station."}, {"id": "blue_gate", "name": "Gate of the Blue Screen", "source_object": "laptop", "description": "A glowing station gate that needs a fresh power cell."}, ], "grounding": [ {"source_object": "charger cable", "world_object": "Black Cable Road", "role": "region", "asset_key": "path_cable"}, {"source_object": "battery pack", "world_object": "Power Cell", "role": "key item", "asset_key": "battery"}, {"source_object": "laptop", "world_object": "Gate of the Blue Screen", "role": "quest goal", "asset_key": "portal"}, ], "npcs": [ { "id": "patchbot", "name": "Patchbot V7", "x": 8, "y": 1, "sprite_key": "npc_robot", "role": "guide", "dialogue": "The portal is out of power. Bring it the Power Cell.", } ], "items": [ { "id": "power_cell", "name": "Power Cell", "x": 4, "y": 5, "sprite_key": "battery", "description": "A humming cell scavenged from a desk gadget.", } ], "quest": { "goal": "Recover the Power Cell and reboot the portal.", "required_item": "power_cell", "success_ending": "The portal comes online and the station lights stabilize.", }, }, { "title": "Archive Garden", "genre": "haunted library mystery", "theme": "haunted_mystery", "tile_palette": {"W": "wall_stone", ".": "floor_stone", "G": "door"}, "player_sprite_key": "player", "tiles": [ "WWWWWWWWWWWW", "W....W.....W", "W....W.....W", "W..........W", "W..WWWW....W", "W..........W", "W......G...W", "WWWWWWWWWWWW", ], "player_start": [1, 2], "regions": [ {"id": "paperleaf_walk", "name": "Paperleaf Walk", "source_object": "open notebook", "description": "A path lined with leaves that rustle like turning pages."}, {"id": "index_gate", "name": "Index Gate", "source_object": "bookmark", "description": "A quiet gate marked with a blank index card."}, ], "grounding": [ {"source_object": "open notebook", "world_object": "Paperleaf Walk", "role": "region", "asset_key": "book"}, {"source_object": "pen cap", "world_object": "Catalog Seal", "role": "key item", "asset_key": "note"}, {"source_object": "bookmark", "world_object": "Index Gate", "role": "quest goal", "asset_key": "door"}, ], "npcs": [ { "id": "archivist", "name": "Archivist Vale", "x": 7, "y": 1, "sprite_key": "npc_librarian", "role": "lorekeeper", "dialogue": "Every gate in this garden wants a citation. Find the Catalog Seal.", } ], "items": [ { "id": "catalog_seal", "name": "Catalog Seal", "x": 3, "y": 5, "sprite_key": "note", "description": "A neat wax seal marked with a shelf number.", } ], "quest": { "goal": "Bring the Catalog Seal to the Index Gate.", "required_item": "catalog_seal", "success_ending": "The Index Gate files itself open, revealing the hidden reading room.", }, }, { "title": "Cableblock Crossing", "genre": "tiny city campus run", "theme": "tiny_city", "tile_palette": {"W": "wall_brick", ".": "floor_city", "G": "sign"}, "player_sprite_key": "player", "tiles": [ "WWWWWWWWWWWW", "W..........W", "W..W..W....W", "W..W..W....W", "W..........W", "W....WW....W", "W......G...W", "WWWWWWWWWWWW", ], "player_start": [1, 1], "regions": [ {"id": "sidewalk_loop", "name": "Sidewalk Loop", "source_object": "charging cable", "description": "A blocky crossing shaped like a cable on a desk."}, {"id": "notice_gate", "name": "Notice Gate", "source_object": "calendar", "description": "A campus sign that opens after the missing coin is found."}, ], "grounding": [ {"source_object": "charging cable", "world_object": "Sidewalk Loop", "role": "region", "asset_key": "path_cable"}, {"source_object": "coin jar", "world_object": "Transit Coin", "role": "key item", "asset_key": "coin"}, {"source_object": "calendar", "world_object": "Notice Gate", "role": "quest goal", "asset_key": "sign"}, ], "npcs": [ { "id": "crossing_guard", "name": "Crossing Guard Mira", "x": 5, "y": 2, "sprite_key": "npc_citizen", "role": "blocker", "dialogue": "The gate is waiting for the Transit Coin. Check the plaza path.", } ], "items": [ { "id": "transit_coin", "name": "Transit Coin", "x": 3, "y": 4, "sprite_key": "coin", "description": "A small token with a map scratched into the edge.", } ], "quest": { "goal": "Find the Transit Coin and open the Notice Gate.", "required_item": "transit_coin", "success_ending": "The Notice Gate flips open and the campus path clears.", }, }, ] SAMPLE_WORLD_LOOKUP = {world["title"]: world for world in SAMPLE_WORLDS} CUSTOM_CSS = """ .pw-note { color: #64748b; font-size: 0.98rem; margin-bottom: 0.75rem; } .pw-ai-box { border: 1px solid #dbe3ef; border-radius: 8px; padding: 14px; margin: 12px 0 16px; background: #f8fafc; } .pw-ai-box h2 { margin-top: 0; } .pw-ai-muted { color: #64748b; font-size: 0.94rem; } .pw-side-card { --pw-panel-bg: #f8fafc; --pw-panel-text: #0f172a; --pw-panel-muted: #334155; --pw-panel-border: #dbe3ef; --pw-panel-rule: #e2e8f0; border: 1px solid var(--pw-panel-border); border-radius: 8px; padding: 14px; background: var(--pw-panel-bg); color: var(--pw-panel-text); overflow-x: hidden; margin-bottom: 12px; } .pw-side-card.pw-theme-dark { --pw-panel-bg: #111827; --pw-panel-text: #e5e7eb; --pw-panel-muted: #cbd5e1; --pw-panel-border: #334155; --pw-panel-rule: #1f2937; } @media (prefers-color-scheme: dark) { .pw-note { color: #cbd5e1; } .pw-ai-box { border-color: #334155; background: #111827; } .pw-ai-muted { color: #cbd5e1; } .pw-side-card.pw-theme-auto { --pw-panel-bg: #111827; --pw-panel-text: #e5e7eb; --pw-panel-muted: #cbd5e1; --pw-panel-border: #334155; --pw-panel-rule: #1f2937; } } .pw-side-card h3 { margin: 0 0 10px; color: var(--pw-panel-text); font-size: 1rem; } .pw-side-card table { width: 100%; min-width: 0; table-layout: fixed; border-collapse: collapse; font-size: 0.9rem; } .pw-side-card th:nth-child(1), .pw-side-card td:nth-child(1) { width: 34%; } .pw-side-card th:nth-child(2), .pw-side-card td:nth-child(2) { width: 42%; } .pw-side-card th:nth-child(3), .pw-side-card td:nth-child(3) { width: 24%; } .pw-side-card th, .pw-side-card td { border-top: 1px solid var(--pw-panel-rule); padding: 8px 6px; text-align: left; vertical-align: top; overflow-wrap: anywhere; } .pw-side-card th, .pw-score-label { color: var(--pw-panel-muted); font-weight: 700; } .pw-side-card td, .pw-side-card p { color: var(--pw-panel-text); } .pw-side-card code { white-space: normal; } .pw-score-grid { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 10px; } .pw-score-item { border-top: 1px solid var(--pw-panel-rule); padding-top: 8px; } .pw-score-value { display: block; margin-top: 2px; color: var(--pw-panel-text); } .pw-warning-list { margin: 10px 0 0; padding-left: 18px; } .pw-game-shell { align-items: flex-start; flex-wrap: wrap; } @media (max-width: 780px) { .pw-game-shell { flex-direction: column !important; } .pw-game-shell > * { width: 100% !important; min-width: 0 !important; } } @media (max-width: 720px) { .pw-score-grid { grid-template-columns: 1fr; } } """ _ASSET_CATALOG_CACHE = None def _is_int_pair(value): return ( isinstance(value, list) and len(value) == 2 and all(isinstance(coord, int) and not isinstance(coord, bool) for coord in value) ) def _clean_object_name(value): text = str(value).strip() if value is not None else "" return text or "mystery object" def _slugify(text): slug_chars = [] previous_was_separator = False for char in _clean_object_name(text).lower(): if char.isalnum(): slug_chars.append(char) previous_was_separator = False elif not previous_was_separator: slug_chars.append("_") previous_was_separator = True slug = "".join(slug_chars).strip("_") return slug or "object" def normalize_theme(display_theme): if display_theme not in THEME_OPTIONS: return DEFAULT_THEME return display_theme def _theme_class(display_theme): return "pw-theme-" + normalize_theme(display_theme).lower() def load_asset_catalog(): global _ASSET_CATALOG_CACHE if _ASSET_CATALOG_CACHE is not None: return copy.deepcopy(_ASSET_CATALOG_CACHE) if ASSET_CATALOG_PATH.exists(): with ASSET_CATALOG_PATH.open("r", encoding="utf-8") as handle: catalog = json.load(handle) else: catalog = copy.deepcopy(EMBEDDED_ASSET_CATALOG) catalog = normalize_asset_catalog(catalog) validate_asset_catalog(catalog) _ASSET_CATALOG_CACHE = catalog return copy.deepcopy(catalog) def normalize_asset_catalog(catalog): normalized = copy.deepcopy(catalog) if "themes" in normalized: return normalized theme_defaults = normalized.get("theme_defaults", {}) themes = {} for theme_name, theme_data in theme_defaults.items(): if not isinstance(theme_data, dict): continue themes[theme_name] = { "tile_palette": theme_data.get("tile_palette", {}), "player_sprite_key": theme_data.get( "player_sprite_key", theme_data.get("player", "player") ), "npc_sprite_keys": theme_data.get( "npc_sprite_keys", theme_data.get("npcs", []) ), "item_sprite_keys": theme_data.get( "item_sprite_keys", theme_data.get("items", []) ), "landmark_asset_keys": theme_data.get( "landmark_asset_keys", theme_data.get("landmarks", []) ), } normalized["themes"] = themes return normalized def _flatten_asset_catalog(catalog): flat = {} for category, entries in catalog.get("assets", {}).items(): if not isinstance(entries, dict): continue for key, entry in entries.items(): if isinstance(entry, dict): flat[key] = {**entry, "key": key, "category": category} return flat def _asset_file_exists(entry): path = entry.get("path") if isinstance(entry, dict) else None return bool(path) and (PROJECT_ROOT / path).exists() def validate_asset_catalog(catalog): if not isinstance(catalog, dict): raise ValueError("asset catalog must be an object") if catalog.get("schema_version") != ASSET_SCHEMA_VERSION: raise ValueError("asset catalog schema_version is unsupported") assets = catalog.get("assets") themes = catalog.get("themes") if not isinstance(assets, dict) or not isinstance(themes, dict): raise ValueError("asset catalog must include assets and themes") for category in ("tiles", "chars", "items", "landmarks"): if not isinstance(assets.get(category), dict) or not assets[category]: raise ValueError(f"asset catalog must include non-empty {category}") for key, entry in assets[category].items(): if not isinstance(entry, dict) or not isinstance(entry.get("path"), str): raise ValueError(f"asset {key} must include a path") if not entry["path"].endswith(".png"): raise ValueError(f"asset {key} must be a PNG path") flat = _flatten_asset_catalog(catalog) for theme in WORLD_THEMES: if theme not in themes: raise ValueError(f"asset catalog missing theme {theme}") theme_data = themes[theme] palette = theme_data.get("tile_palette", {}) for symbol in ("W", ".", "G"): if palette.get(symbol) not in flat: raise ValueError(f"theme {theme} has invalid tile asset for {symbol}") if theme_data.get("player_sprite_key") not in flat: raise ValueError(f"theme {theme} has invalid player sprite") return True def _catalog_entry(catalog, asset_key): return _flatten_asset_catalog(catalog).get(asset_key) def _is_known_asset(catalog, asset_key): return _catalog_entry(catalog, asset_key) is not None def _category_has_asset(catalog, category, asset_key): return asset_key in catalog.get("assets", {}).get(category, {}) def normalize_world(world): if not isinstance(world, dict): raise ValueError("World must be a JSON object.") normalized = copy.deepcopy(world) normalized.setdefault("schema_version", WORLD_SCHEMA_VERSION) normalized.setdefault("genre", "pocket adventure") normalized.setdefault("regions", []) normalized.setdefault("grounding", []) normalized.setdefault("npcs", []) normalized.setdefault("items", []) normalized.setdefault("landmarks", []) normalized.setdefault("source", {"kind": "sample", "objects": []}) normalized.setdefault("theme", DEFAULT_WORLD_THEME) normalized.setdefault("intro", f"A tiny playable world called {normalized.get('title', 'PocketWorld')}.") tiles = normalized.get("tiles") if isinstance(tiles, list): tile_aliases = {"Q": "G", "P": ".", "N": ".", "I": " "} normalized["tiles"] = [ "".join(tile_aliases.get(char, char) for char in row).replace(" ", ".") if isinstance(row, str) else row for row in tiles ] return normalized def validate_detected_objects(detected_objects): errors = [] if not isinstance(detected_objects, dict): raise ValueError("Detected objects payload must be a JSON object.") objects = detected_objects.get("objects") if not isinstance(objects, list): errors.append("objects must be a list") else: for index, obj in enumerate(objects): label = f"objects[{index}]" if not isinstance(obj, dict): errors.append(f"{label} must be an object") continue if not isinstance(obj.get("label"), str) or not obj.get("label").strip(): errors.append(f"{label}.label must be a non-empty string") confidence = obj.get("confidence") if confidence is not None and not isinstance(confidence, (int, float)): errors.append(f"{label}.confidence must be a number when provided") bbox = obj.get("bbox") if bbox is not None and ( not isinstance(bbox, list) or len(bbox) != 4 or not all(isinstance(value, (int, float)) for value in bbox) ): errors.append(f"{label}.bbox must be [x, y, width, height] when provided") if errors: raise ValueError("; ".join(errors)) return True def _infer_grounding_asset(entry, world, catalog): role = str(entry.get("role", "")).lower() if "goal" in role or "gate" in role: return world["tile_palette"]["G"] if "item" in role: required_item = world.get("quest", {}).get("required_item") for item in world.get("items", []): if item.get("id") == required_item: return item.get("sprite_key", "key") return "key" if "region" in role: return world["tile_palette"]["."] theme_data = catalog["themes"][world["theme"]] return theme_data["landmark_asset_keys"][0] def _step_id(step_type, target, index): return f"{step_type}_{target or index + 1}".replace(" ", "_") def _step_text(step_type, target, world): if step_type == "talk": npc = next((npc for npc in world.get("npcs", []) if npc.get("id") == target), None) return f"Talk to {npc.get('name', 'the guide')}." if npc else "Talk to the guide." if step_type == "inspect": landmark = next((landmark for landmark in world.get("landmarks", []) if landmark.get("id") == target), None) return f"Inspect {landmark.get('name', 'the landmark')}." if landmark else "Inspect the landmark." if step_type == "collect": item = next((item for item in world.get("items", []) if item.get("id") == target), None) return f"Find {item.get('name', 'the required item')}." if item else "Find the required item." if step_type == "unlock": return "Unlock the final gate." return "Continue the quest." def _default_quest_steps(world): steps = [] if world.get("npcs"): target = world["npcs"][0].get("id", "guide") steps.append({"id": _step_id("talk", target, 0), "type": "talk", "target": target}) required_item = world.get("quest", {}).get("required_item") if required_item: steps.append({"id": _step_id("collect", required_item, len(steps)), "type": "collect", "target": required_item}) goal_target = world.get("quest", {}).get("goal_id") or "goal" steps.append({"id": _step_id("unlock", goal_target, len(steps)), "type": "unlock", "target": goal_target}) return steps def _normalize_quest_steps(world): raw_steps = world.get("quest_steps") if not isinstance(raw_steps, list) or not raw_steps: raw_steps = _default_quest_steps(world) steps = [] for index, step in enumerate(raw_steps): if not isinstance(step, dict): continue step_type = step.get("type", "") target = step.get("target", "") normalized_step = { "id": step.get("id") or _step_id(step_type, target, index), "step": step.get("step", index + 1), "type": step_type, "target": target, "text": step.get("text") or _step_text(step_type, target, world), } steps.append(normalized_step) if not steps: steps = _default_quest_steps(world) for index, step in enumerate(steps): step["step"] = index + 1 step["text"] = _step_text(step["type"], step["target"], world) return steps def normalize_world_assets(world, catalog=None): catalog = catalog or load_asset_catalog() normalized = normalize_world(world) warnings = [] theme = normalized.get("theme", DEFAULT_WORLD_THEME) if theme not in catalog["themes"]: warnings.append(f"Unknown theme '{theme}' replaced with '{DEFAULT_WORLD_THEME}'.") theme = DEFAULT_WORLD_THEME normalized["theme"] = theme theme_data = catalog["themes"][theme] palette = copy.deepcopy(theme_data["tile_palette"]) palette.update(normalized.get("tile_palette", {})) for symbol in ("W", ".", "G"): candidate = palette.get(symbol) expected_category = "landmarks" if symbol == "G" else "tiles" if not _category_has_asset(catalog, expected_category, candidate): fallback = theme_data["tile_palette"][symbol] warnings.append(f"Unknown tile asset '{candidate}' for '{symbol}' replaced with '{fallback}'.") palette[symbol] = fallback normalized["tile_palette"] = palette player_key = normalized.get("player_sprite_key", theme_data["player_sprite_key"]) if not _category_has_asset(catalog, "chars", player_key): warnings.append(f"Unknown player sprite '{player_key}' replaced with '{theme_data['player_sprite_key']}'.") player_key = theme_data["player_sprite_key"] normalized["player_sprite_key"] = player_key npc_defaults = theme_data.get("npc_sprite_keys", ["npc_citizen"]) for index, npc in enumerate(normalized.get("npcs", [])): fallback = npc_defaults[index % len(npc_defaults)] sprite_key = npc.get("sprite_key", fallback) if not _category_has_asset(catalog, "chars", sprite_key): warnings.append(f"Unknown NPC sprite '{sprite_key}' on {npc.get('id', index)} replaced with '{fallback}'.") sprite_key = fallback npc["sprite_key"] = sprite_key npc.setdefault("role", "guide" if index == 0 else "lorekeeper") item_defaults = theme_data.get("item_sprite_keys", ["key"]) for index, item in enumerate(normalized.get("items", [])): fallback = item_defaults[index % len(item_defaults)] sprite_key = item.get("sprite_key", fallback) if not _category_has_asset(catalog, "items", sprite_key): warnings.append(f"Unknown item sprite '{sprite_key}' on {item.get('id', index)} replaced with '{fallback}'.") sprite_key = fallback item["sprite_key"] = sprite_key landmark_defaults = theme_data.get("landmark_asset_keys", ["gate"]) for index, landmark in enumerate(normalized.get("landmarks", [])): fallback = landmark_defaults[index % len(landmark_defaults)] sprite_key = landmark.get("sprite_key", fallback) if not _category_has_asset(catalog, "landmarks", sprite_key): warnings.append(f"Unknown landmark sprite '{sprite_key}' on {landmark.get('id', index)} replaced with '{fallback}'.") sprite_key = fallback landmark["sprite_key"] = sprite_key landmark.setdefault("source_object", "unknown source object") landmark.setdefault("description", "A grounded landmark in this pocket world.") for grounding in normalized.get("grounding", []): asset_key = grounding.get("asset_key") or _infer_grounding_asset(grounding, normalized, catalog) if not _is_known_asset(catalog, asset_key): fallback = _infer_grounding_asset(grounding, normalized, catalog) warnings.append(f"Unknown grounding asset '{asset_key}' replaced with '{fallback}'.") asset_key = fallback grounding["asset_key"] = asset_key normalized["quest_steps"] = _normalize_quest_steps(normalized) asset_warnings = validate_asset_usage(normalized, catalog) normalized["asset_warnings"] = warnings + asset_warnings return normalized def _world_asset_refs(world): refs = [] for symbol, asset_key in world.get("tile_palette", {}).items(): refs.append((f"tile_palette.{symbol}", asset_key)) refs.append(("player_sprite_key", world.get("player_sprite_key"))) for npc in world.get("npcs", []): refs.append((f"npcs.{npc.get('id', 'unknown')}.sprite_key", npc.get("sprite_key"))) for item in world.get("items", []): refs.append((f"items.{item.get('id', 'unknown')}.sprite_key", item.get("sprite_key"))) for landmark in world.get("landmarks", []): refs.append((f"landmarks.{landmark.get('id', 'unknown')}.sprite_key", landmark.get("sprite_key"))) for grounding in world.get("grounding", []): refs.append((f"grounding.{grounding.get('world_object', 'unknown')}.asset_key", grounding.get("asset_key"))) return [(where, key) for where, key in refs if key] def validate_asset_usage(world, catalog=None): catalog = catalog or load_asset_catalog() flat = _flatten_asset_catalog(catalog) warnings = [] for where, asset_key in _world_asset_refs(world): entry = flat.get(asset_key) if entry is None: warnings.append(f"{where} references unknown asset '{asset_key}'.") elif not _asset_file_exists(entry): warnings.append(f"{where} asset '{asset_key}' is missing file {entry.get('path')}; using fallback shape.") return warnings def make_seed_world_from_objects(detected_objects, title="Generated PocketWorld"): validate_detected_objects(detected_objects) object_names = [ _clean_object_name(obj.get("label")) for obj in detected_objects.get("objects", []) if isinstance(obj, dict) ] while len(object_names) < 4: object_names.append(["keepsake", "lantern", "gate", "guide"][len(object_names)]) item_id = _slugify(object_names[1] + "_charm") seed = { "title": title, "genre": "model-ready generated micro-world", "theme": DEFAULT_WORLD_THEME, "source": {"kind": "image_objects", "objects": detected_objects.get("objects", [])}, "tiles": [ "WWWWWWWWWWWW", "W..........W", "W..W.......W", "W..W..W....W", "W.....W....W", "W..........W", "W......G...W", "WWWWWWWWWWWW", ], "player_start": [1, 6], "regions": [ { "id": _slugify(object_names[0]), "name": object_names[0].title() + " Crossing", "source_object": object_names[0], "description": "A walkable region grounded in an object detected from the image.", } ], "grounding": [ {"source_object": object_names[0], "world_object": object_names[0].title() + " Crossing", "role": "landmark", "asset_key": "well"}, {"source_object": object_names[1], "world_object": object_names[1].title() + " Charm", "role": "key item", "asset_key": "key"}, {"source_object": object_names[2], "world_object": object_names[2].title() + " Gate", "role": "quest goal", "asset_key": "gate"}, {"source_object": object_names[3], "world_object": object_names[3].title() + " Guide", "role": "npc", "asset_key": "npc_wizard"}, ], "npcs": [ { "id": _slugify(object_names[3] + "_guide"), "name": object_names[3].title() + " Guide", "x": 2, "y": 2, "sprite_key": "npc_wizard", "role": "guide", "dialogue": f"Find the {object_names[1].title()} Charm, then open the gate.", } ], "items": [ { "id": item_id, "name": object_names[1].title() + " Charm", "x": 5, "y": 4, "sprite_key": "key", "description": "A key item improvised from the image object list.", } ], "quest": { "goal": f"Collect the {object_names[1].title()} Charm and reach the gate.", "required_item": item_id, "success_ending": "The generated world opens its final path.", }, } return normalize_world_assets(seed) def _is_reachable(tiles, start, target, allow_goal=False): start_tuple = tuple(start) target_tuple = tuple(target) width = len(tiles[0]) height = len(tiles) queue = [start_tuple] seen = {start_tuple} while queue: x, y = queue.pop(0) if (x, y) == target_tuple: return True for dx, dy in ((1, 0), (-1, 0), (0, 1), (0, -1)): nx = x + dx ny = y + dy if (nx, ny) in seen or not (0 <= nx < width and 0 <= ny < height): continue tile = tiles[ny][nx] if tile == "W" or (tile == "G" and not allow_goal and (nx, ny) != target_tuple): continue seen.add((nx, ny)) queue.append((nx, ny)) return False def validate_world(world): errors = [] if not isinstance(world, dict): raise ValueError("World must be a JSON object.") world = normalize_world(world) tiles = world.get("tiles") if not isinstance(tiles, list) or not tiles: errors.append("tiles must be a non-empty list of strings") width = 0 height = 0 elif not all(isinstance(row, str) and row for row in tiles): errors.append("each tile row must be a non-empty string") width = 0 height = len(tiles) else: width = len(tiles[0]) height = len(tiles) if any(len(row) != width for row in tiles): errors.append("all tile rows must have the same length") invalid_chars = sorted({char for row in tiles for char in row} - set(TILE_LEGEND.keys())) if invalid_chars: errors.append(f"tiles contain unsupported symbols: {', '.join(invalid_chars)}") def inside_map(x, y): return width > 0 and height > 0 and 0 <= x < width and 0 <= y < height def passable(x, y): return inside_map(x, y) and tiles[y][x] != "W" player_start = world.get("player_start") if not _is_int_pair(player_start): errors.append("player_start must be [x, y] with integer coordinates") elif not inside_map(player_start[0], player_start[1]): errors.append("player_start must be inside the map") elif not passable(player_start[0], player_start[1]): errors.append("player_start cannot be placed on a wall tile") for collection_name in ("npcs", "items", "landmarks"): collection = world.get(collection_name, []) if not isinstance(collection, list): errors.append(f"{collection_name} must be a list") continue for index, entity in enumerate(collection): label = f"{collection_name}[{index}]" if not isinstance(entity, dict): errors.append(f"{label} must be an object") continue if not isinstance(entity.get("id"), str) or not entity.get("id").strip(): errors.append(f"{label}.id must be a non-empty string") if not isinstance(entity.get("name"), str) or not entity.get("name").strip(): errors.append(f"{label}.name must be a non-empty string") if collection_name == "landmarks" and not isinstance(entity.get("description"), str): errors.append(f"{label}.description must be a string") x = entity.get("x") y = entity.get("y") if not isinstance(x, int) or isinstance(x, bool) or not isinstance(y, int) or isinstance(y, bool): errors.append(f"{label} must have integer x/y coordinates") elif not inside_map(x, y): errors.append(f"{label} x/y must be inside the map") elif not passable(x, y): errors.append(f"{label} cannot be placed on a wall tile") goal_positions = [] if width > 0 and height > 0: for y, row in enumerate(tiles): for x, tile in enumerate(row): if tile == "G": goal_positions.append([x, y]) if not goal_positions: errors.append("tiles must include at least one G goal tile") items = world.get("items", []) npcs = world.get("npcs", []) landmarks = world.get("landmarks", []) npc_ids = [npc.get("id") for npc in npcs if isinstance(npc, dict)] item_ids = [item.get("id") for item in items if isinstance(item, dict)] landmark_ids = [landmark.get("id") for landmark in landmarks if isinstance(landmark, dict)] if len(npc_ids) != len(set(npc_ids)): errors.append("npc ids must be unique") if len(item_ids) != len(set(item_ids)): errors.append("item ids must be unique") if len(landmark_ids) != len(set(landmark_ids)): errors.append("landmark ids must be unique") quest = world.get("quest") goal_id = "goal" if not isinstance(quest, dict): errors.append("quest must be an object") else: goal_id = quest.get("goal_id") or "goal" if goal_id is not None and not isinstance(goal_id, str): errors.append("quest.goal_id must be a string when provided") required_item = quest.get("required_item") if not isinstance(required_item, str) or not required_item: errors.append("quest.required_item must be a non-empty item id") elif required_item not in item_ids: errors.append("quest.required_item must match one item id") elif width > 0 and height > 0 and _is_int_pair(player_start): required_item_obj = next( (item for item in items if isinstance(item, dict) and item.get("id") == required_item), None, ) if required_item_obj and not _is_reachable( tiles, player_start, [required_item_obj["x"], required_item_obj["y"]], allow_goal=False, ): errors.append("quest.required_item must be reachable from player_start") if required_item_obj and goal_positions and not any( _is_reachable( tiles, [required_item_obj["x"], required_item_obj["y"]], goal, allow_goal=True, ) for goal in goal_positions ): errors.append("a G goal tile must be reachable after collecting quest.required_item") steps = world.get("quest_steps", []) if steps is not None: if not isinstance(steps, list): errors.append("quest_steps must be a list when provided") else: seen_step_ids = set() allowed_step_types = {"talk", "inspect", "collect", "unlock"} for index, step in enumerate(steps): label = f"quest_steps[{index}]" if not isinstance(step, dict): errors.append(f"{label} must be an object") continue step_id = step.get("id") step_type = step.get("type") target = step.get("target") if not isinstance(step_id, str) or not step_id.strip(): errors.append(f"{label}.id must be a non-empty string") elif step_id in seen_step_ids: errors.append(f"{label}.id must be unique") else: seen_step_ids.add(step_id) if step_type not in allowed_step_types: errors.append(f"{label}.type must be one of talk, inspect, collect, unlock") continue if not isinstance(target, str) or not target.strip(): errors.append(f"{label}.target must be a non-empty string") continue if step_type == "talk" and target not in npc_ids: errors.append(f"{label}.target must match an NPC id") elif step_type == "inspect" and target not in landmark_ids: errors.append(f"{label}.target must match a landmark id") elif step_type == "collect" and target not in item_ids: errors.append(f"{label}.target must match an item id") elif step_type == "unlock" and target not in {"goal", goal_id}: errors.append(f"{label}.target must be 'goal' or quest.goal_id") if step.get("text") is not None and not isinstance(step.get("text"), str): errors.append(f"{label}.text must be a string when provided") if errors: raise ValueError("; ".join(errors)) return True def _playability_ok(world): try: validate_world(world) return True except ValueError: return False def compute_world_score(world, catalog=None): catalog = catalog or load_asset_catalog() world = normalize_world_assets(world, catalog) refs = _world_asset_refs(world) flat = _flatten_asset_catalog(catalog) valid_refs = 0 for _, key in refs: entry = flat.get(key) if entry is not None and _asset_file_exists(entry): valid_refs += 1 grounding_total = len(world.get("source", {}).get("objects") or world.get("grounding", [])) grounding_used = len( { entry.get("source_object") for entry in world.get("grounding", []) if entry.get("source_object") and entry.get("world_object") } ) required_item = world.get("quest", {}).get("required_item") item_ids = {item.get("id") for item in world.get("items", [])} steps = world.get("quest_steps", []) quest_ok = bool(required_item in item_ids and any(step.get("type") == "collect" for step in steps) and any(step.get("type") == "unlock" for step in steps)) return { "theme": world.get("theme"), "grounding": {"used": grounding_used, "total": max(grounding_total, 1)}, "playability": _playability_ok(world), "quest_complete": quest_ok, "asset_validity": {"valid": valid_refs, "total": max(len(refs), 1)}, "warnings": world.get("asset_warnings", []), } def _truthy_env(name): return os.getenv(name, "").strip().lower() in {"1", "true", "yes", "on"} def _float_env(name, default): try: return float(os.getenv(name, default)) except (TypeError, ValueError): return float(default) def _debug_log(message, payload=None): prefix = "[PocketWorld Phase 4A]" if payload is None: print(f"{prefix} {message}", flush=True) return try: rendered_payload = json.dumps(payload, ensure_ascii=True, default=str)[:2000] except Exception: rendered_payload = str(payload)[:2000] print(f"{prefix} {message}: {rendered_payload}", flush=True) def _preview_text(value, limit=500): if value is None: return "" text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=True, default=str) return text.replace("\n", "\\n")[:limit] def _allowed_asset_keys(catalog): return { category: sorted(entries.keys()) for category, entries in catalog.get("assets", {}).items() if isinstance(entries, dict) } def _scene_source_objects(scene_parse): return [ { "id": _slugify(obj.get("name", f"object_{index + 1}")), "label": obj.get("name", f"object {index + 1}"), "confidence": 1.0, "attributes": { "location": obj.get("location", ""), "role_hint": obj.get("role_hint", ""), }, } for index, obj in enumerate(scene_parse.get("objects", [])) if isinstance(obj, dict) ] def _prompt_summary(scene_parse, genre_label, tone_label, template_label, world_size, catalog): allowed_assets = _allowed_asset_keys(catalog) return { "model": TEXT_WORLD_MODEL_ID, "source_scene": scene_parse.get("source_scene"), "scene_type": scene_parse.get("scene_type"), "genre": genre_label, "tone": tone_label, "game_template": template_label, "world_size": world_size, "target_theme": AI_GENRE_CONFIG.get(genre_label, AI_GENRE_CONFIG["Cozy Fantasy"])["theme"], "allowed_asset_counts": { category: len(keys) for category, keys in allowed_assets.items() }, "required_top_level_keys": [ "title", "intro", "genre", "theme", "tile_palette", "tiles", "player_start", "grounding", "regions", "npcs", "items", "landmarks", "quest", "quest_steps", ], } def _schema_rules_for_prompt(catalog): allowed_assets = _allowed_asset_keys(catalog) return { "world_schema": WORLD_SCHEMA, "allowed_asset_keys": allowed_assets, "valid_themes": WORLD_THEMES, "valid_tile_symbols": sorted(TILE_LEGEND.keys()), "valid_quest_step_types": ["talk", "inspect", "collect", "unlock"], "strict_rules": [ "Output one valid JSON object only.", "Do not include markdown fences or prose.", "Use only W, ., and G in tile rows.", "All tile rows must have the same length.", "Coordinates use [x, y] with x from the left and y from the top.", "Place player_start, NPCs, items, landmarks, and G on passable coordinates.", "Use only allowed asset keys in tile_palette, player_sprite_key, sprite_key, and asset_key fields.", "required_item must exactly match one item id.", "quest step targets must exist: talk targets NPC id, inspect targets landmark id, collect targets item id, unlock targets quest.goal_id or goal.", "The player must be able to reach the required item, then reach a G gate.", "Include at least 2 NPCs, 2 landmarks, 2 items, and 4 grounding entries.", ], } def build_world_generation_prompt(scene_parse, genre_label, tone_label, template_label, world_size, catalog): genre_config = AI_GENRE_CONFIG.get(genre_label, AI_GENRE_CONFIG["Cozy Fantasy"]) width_hint = "24 columns x 16 rows" if world_size == "Large" else "20 columns x 14 rows" prompt_payload = { "task": "Generate a PocketWorld Studio playable world JSON for the existing Phaser renderer.", "source_scene_parse": scene_parse, "selected_genre": genre_label, "selected_tone": tone_label, "selected_game_template": template_label, "template_rule": AI_TEMPLATE_RULES.get(template_label, AI_TEMPLATE_RULES["Quest Gate"]), "world_size": world_size, "map_size_hint": width_hint, "target_theme": genre_config["theme"], "target_genre": genre_config["genre"], "schema_and_rules": _schema_rules_for_prompt(catalog), "design_requirements": [ "Make an explorable micro-world with paths, walls, a starting area, NPC area, landmark area, item area, and final locked gate.", "Use scene objects as grounded transformations, not literal real-world labels only.", "Make dialogue concise and useful.", "Include quest_steps in this order when possible: talk, inspect, collect, unlock.", "The final G tile should represent the locked gate or final exit.", "Return JSON only.", ], } return ( "You are PocketWorld Studio's world JSON generator. You do not write Phaser code. " "You only produce validated world JSON for a renderer that already exists.\n\n" + json.dumps(prompt_payload, ensure_ascii=True, indent=2) ) def build_compact_world_generation_prompt(scene_parse, genre_label, tone_label, template_label, world_size, catalog): genre_config = AI_GENRE_CONFIG.get(genre_label, AI_GENRE_CONFIG["Cozy Fantasy"]) theme = genre_config["theme"] theme_data = catalog["themes"][theme] map_size = "24x16" if world_size == "Large" else "20x14" seed_world = make_fake_ai_world( scene_parse, genre_label, tone_label, template_label, world_size, catalog, ) seed_world["title"] = f"Generated AI World: {scene_parse.get('source_scene', 'Scene')}" seed_world["intro"] = ( f"A {tone_label.lower()} {genre_label.lower()} pocket world generated from " f"{scene_parse.get('source_scene', 'the source scene')}." ) payload = { "instruction": ( "Return the JSON object only. No markdown, no prose, no , no analysis. " "Copy the provided complete object and only improve names, descriptions, and dialogue if needed." ), "scene": scene_parse, "selected_controls": { "genre": genre_label, "tone": tone_label, "template": template_label, "required_map_size": map_size, "required_theme": theme, "template_rule": AI_TEMPLATE_RULES.get(template_label, AI_TEMPLATE_RULES["Quest Gate"]), }, "hard_rules": [ "The first character of the response must be {.", "The last character of the response must be }.", "Keep every id, coordinate, tile row, required_item, and quest step target valid.", "Use only these tile symbols: W . G.", "Use only these tile palette keys: " + ", ".join(theme_data["tile_palette"].values()), "Use only these NPC sprite keys: " + ", ".join(theme_data.get("npc_sprite_keys", [])), "Use only these item sprite keys: " + ", ".join(theme_data.get("item_sprite_keys", [])), "Use only these landmark sprite keys: " + ", ".join(theme_data.get("landmark_asset_keys", [])), ], "complete_valid_object_to_copy": seed_world, } return ( "You complete JSON for PocketWorld Studio. Do not reason. Do not output . " "Return one valid JSON object and nothing else. The first character must be { and the last must be }.\n\n" + json.dumps(payload, ensure_ascii=True, indent=2) ) def build_repair_prompt(invalid_payload, validation_errors, scene_parse, genre_label, tone_label, template_label, world_size, catalog): repair_payload = { "task": "Repair this PocketWorld world JSON so it validates.", "invalid_world_or_raw_output": invalid_payload, "validation_errors": validation_errors, "source_scene_parse": scene_parse, "selected_genre": genre_label, "selected_tone": tone_label, "selected_game_template": template_label, "world_size": world_size, "schema_and_rules": _schema_rules_for_prompt(catalog), "repair_rules": [ "Return one corrected JSON object only.", "The first character must be { and the last character must be }.", "Do not include markdown fences, explanations, analysis, or .", "Preserve the selected scene and genre as much as possible.", "Fix coordinates, IDs, required_item, quest_steps, tile rows, and reachability.", ], } return ( "You repair PocketWorld Studio world JSON. Output one corrected JSON object only.\n\n" + json.dumps(repair_payload, ensure_ascii=True, indent=2) ) def _scene_object_names(scene_parse): names = [ _clean_object_name(obj.get("name")) for obj in scene_parse.get("objects", []) if isinstance(obj, dict) and obj.get("name") ] while len(names) < 6: names.append(["keepsake", "archive card", "gate light", "road charm", "spare key", "beacon"][len(names)]) return names def _safe_string(value, fallback, limit=90): if isinstance(value, str) and value.strip(): return value.strip()[:limit] return fallback def _safe_string_list(value, fallbacks, limit=4): result = [] if isinstance(value, list): for item in value: if isinstance(item, str) and item.strip(): result.append(item.strip()[:90]) if len(result) >= limit: break for fallback in fallbacks: if len(result) >= limit: break result.append(fallback) return result def validate_design_patch(patch, scene_parse=None): if not isinstance(patch, dict): raise ValueError("design patch must be a JSON object") forbidden_keys = {"instruction", "scene", "task", "invalid_world_or_raw_output", "tiles", "player_start", "npcs", "items", "landmarks", "quest"} present_forbidden = sorted(forbidden_keys.intersection(patch.keys())) if present_forbidden: raise ValueError(f"design patch includes forbidden keys: {', '.join(present_forbidden)}") scene_parse = scene_parse or {} object_names = _scene_object_names(scene_parse) normalized = { "title": _safe_string(patch.get("title"), f"Generated PocketWorld: {scene_parse.get('source_scene', 'Scene')}"), "intro": _safe_string( patch.get("intro"), f"A tiny world built from {scene_parse.get('source_scene', 'the source scene')}.", limit=180, ), "world_tone": _safe_string(patch.get("world_tone"), "playful"), "transformations": [], "npc_names": _safe_string_list( patch.get("npc_names"), [f"{object_names[0].title()} Guide", f"{object_names[3].title()} Helper"], limit=2, ), "landmark_names": _safe_string_list( patch.get("landmark_names"), [f"{object_names[1].title()} Archive", f"{object_names[5].title()} Beacon"], limit=2, ), "item_names": _safe_string_list( patch.get("item_names"), [f"{object_names[4].title()} Key", "Optional Spark"], limit=2, ), "quest_lines": _safe_string_list( patch.get("quest_lines"), ["Talk to the guide.", "Inspect the landmark.", "Collect the key.", "Unlock the gate."], limit=4, ), "dialogue_lines": _safe_string_list( patch.get("dialogue_lines"), [ "The route is open, but the gate wants a key.", "Inspect the landmark before you search for the item.", "You are close now.", ], limit=3, ), "ending": _safe_string(patch.get("ending"), "The gate opens and the pocket world settles into place.", limit=160), } transformations = patch.get("transformations", []) if isinstance(transformations, list): for index, entry in enumerate(transformations[:6]): if not isinstance(entry, dict): continue source = _safe_string(entry.get("source_object"), object_names[index % len(object_names)]) normalized["transformations"].append( { "source_object": source, "world_object": _safe_string(entry.get("world_object"), f"{source.title()} World Object"), "role": _safe_string(entry.get("role"), "landmark"), "why": _safe_string(entry.get("why"), f"Inspired by {source}.", limit=140), } ) while len(normalized["transformations"]) < 4: index = len(normalized["transformations"]) source = object_names[index] role = ["landmark", "clue", "gate", "item"][index % 4] normalized["transformations"].append( { "source_object": source, "world_object": f"{source.title()} {role.title()}", "role": role, "why": f"Because {source} can become a tiny game {role}.", } ) return normalized def _template_key(game_template, world_size): size = "large" if world_size == "Large" else "medium" template_slug = { "Quest Gate": "quest_gate", "Mystery Trail": "mystery_trail", "Rescue Route": "rescue_route", }.get(game_template, "quest_gate") return f"{size}_{template_slug}" def _grounding_asset_for_role(role, theme_data, npc_key, item_key, landmark_key): lowered = str(role or "").lower() if any(word in lowered for word in ("gate", "goal", "exit")): return theme_data["tile_palette"]["G"] if "npc" in lowered or "guide" in lowered: return npc_key if "item" in lowered or "key" in lowered or "clue" in lowered: return item_key if "road" in lowered or "path" in lowered or "region" in lowered: return theme_data["tile_palette"]["."] return landmark_key def build_world_from_design_patch( design_patch, scene_parse, genre, tone, game_template, world_size, asset_catalog, ): patch = validate_design_patch(design_patch, scene_parse) genre_config = AI_GENRE_CONFIG.get(genre, AI_GENRE_CONFIG["Cozy Fantasy"]) theme = genre_config["theme"] theme_data = asset_catalog["themes"][theme] template_name = _template_key(game_template, world_size) template = WORLD_BUILDER_TEMPLATES.get(template_name, WORLD_BUILDER_TEMPLATES["large_quest_gate"]) npc_keys = theme_data.get("npc_sprite_keys", ["npc_citizen", "npc_merchant"]) item_keys = theme_data.get("item_sprite_keys", ["key", "gem"]) landmark_keys = theme_data.get("landmark_asset_keys", ["tower", "gate"]) object_names = _scene_object_names(scene_parse) required_item_id = "generated_key" optional_item_id = "optional_token" goal_id = "generated_gate" npc_slots = template["npc_slots"] landmark_slots = template["landmark_slots"] item_slots = template["item_slots"] transformations = patch["transformations"] grounding = [] for index, entry in enumerate(transformations[:6]): grounding.append( { "source_object": entry["source_object"], "world_object": entry["world_object"], "role": entry["role"], "asset_key": _grounding_asset_for_role( entry["role"], theme_data, npc_keys[index % len(npc_keys)], item_keys[index % len(item_keys)], landmark_keys[index % len(landmark_keys)], ), } ) regions = [] for index, region in enumerate(template["regions"]): source = transformations[index % len(transformations)]["source_object"] regions.append( { "id": region[0], "name": transformations[index % len(transformations)].get("world_object") or region[1], "source_object": source, "description": region[2], } ) world = { "title": patch["title"], "genre": genre_config["genre"], "intro": patch["intro"], "theme": theme, "source": { "kind": "model_generated", "generation_mode": "design_patch", "source_scene": scene_parse.get("source_scene"), "scene_type": scene_parse.get("scene_type"), "mood": scene_parse.get("mood"), "objects": _scene_source_objects(scene_parse), }, "tile_palette": theme_data["tile_palette"], "player_sprite_key": theme_data["player_sprite_key"], "tiles": copy.deepcopy(template["tiles"]), "player_start": copy.deepcopy(template["player_start"]), "regions": regions, "grounding": grounding, "npcs": [ { "id": "generated_guide", "name": patch["npc_names"][0], "x": npc_slots[0][0], "y": npc_slots[0][1], "sprite_key": npc_keys[0], "role": "guide", "dialogue": patch["dialogue_lines"][0], }, { "id": "generated_helper", "name": patch["npc_names"][1], "x": npc_slots[1][0], "y": npc_slots[1][1], "sprite_key": npc_keys[1 % len(npc_keys)], "role": "lorekeeper", "dialogue": patch["dialogue_lines"][1], }, ], "items": [ { "id": required_item_id, "name": patch["item_names"][0], "x": item_slots[0][0], "y": item_slots[0][1], "sprite_key": item_keys[0], "description": f"A required item inspired by {object_names[4]}.", }, { "id": optional_item_id, "name": patch["item_names"][1], "x": item_slots[1][0], "y": item_slots[1][1], "sprite_key": item_keys[1 % len(item_keys)], "description": f"An optional charm inspired by {object_names[5]}.", }, ], "landmarks": [ { "id": "generated_landmark", "name": patch["landmark_names"][0], "x": landmark_slots[0][0], "y": landmark_slots[0][1], "sprite_key": landmark_keys[0], "source_object": transformations[1 % len(transformations)]["source_object"], "description": transformations[1 % len(transformations)]["why"], }, { "id": "generated_beacon", "name": patch["landmark_names"][1], "x": landmark_slots[1][0], "y": landmark_slots[1][1], "sprite_key": landmark_keys[1 % len(landmark_keys)], "source_object": transformations[2 % len(transformations)]["source_object"], "description": transformations[2 % len(transformations)]["why"], }, ], "quest": { "goal": " ".join(patch["quest_lines"][:4]), "goal_id": goal_id, "required_item": required_item_id, "success_ending": patch["ending"], }, "quest_steps": [ {"id": "talk_generated_guide", "type": "talk", "target": "generated_guide", "text": patch["quest_lines"][0]}, {"id": "inspect_generated_landmark", "type": "inspect", "target": "generated_landmark", "text": patch["quest_lines"][1]}, {"id": "collect_generated_key", "type": "collect", "target": required_item_id, "text": patch["quest_lines"][2]}, {"id": "unlock_generated_gate", "type": "unlock", "target": goal_id, "text": patch["quest_lines"][3]}, ], } return normalize_world_assets(world, asset_catalog), template_name def make_fake_design_patch(scene_parse, genre_label, tone_label, template_label): object_names = _scene_object_names(scene_parse) return validate_design_patch( { "title": f"Generated Test World: {scene_parse.get('source_scene', 'Scene')}", "intro": f"A fake AI {tone_label.lower()} {genre_label.lower()} design patch for {scene_parse.get('source_scene', 'the scene')}.", "world_tone": tone_label.lower(), "transformations": [ { "source_object": object_names[0], "world_object": f"{object_names[0].title()} Harbor", "role": "landmark", "why": f"{object_names[0]} becomes a grounded landmark.", }, { "source_object": object_names[1], "world_object": f"{object_names[1].title()} Archive", "role": "clue landmark", "why": f"{object_names[1]} suggests a place to inspect.", }, { "source_object": object_names[2], "world_object": f"{object_names[2].title()} Gate", "role": "gate", "why": f"{object_names[2]} becomes the final threshold.", }, { "source_object": object_names[4], "world_object": "Generated Key", "role": "required item", "why": f"{object_names[4]} becomes the key clue.", }, ], "npc_names": [f"{object_names[0].title()} Guide", f"{object_names[3].title()} Helper"], "landmark_names": [f"{object_names[1].title()} Archive", f"{object_names[5].title()} Beacon"], "item_names": ["Generated Key", "Optional Spark"], "quest_lines": [ "Talk to the generated guide.", "Inspect the generated landmark.", "Collect the Generated Key.", "Unlock the generated gate.", ], "dialogue_lines": [ "This fake design patch proves the compact AI pipeline is running.", "The template already knows where everything belongs.", "Bring the key to the gate.", ], "ending": "The fake design patch becomes a playable validated world.", }, scene_parse, ) def build_design_patch_prompt(scene_parse, genre_label, tone_label, template_label): object_names = _scene_object_names(scene_parse)[:6] scene_summary = f"{scene_parse.get('source_scene', 'Scene')} ({scene_parse.get('mood', 'mixed mood')})" schema_keys = ", ".join(DESIGN_PATCH_SCHEMA.keys()) return ( "You create tiny game flavor JSON for PocketWorld Studio.\n" f"Return exactly one compact JSON object with these keys: {schema_keys}.\n" f"Source scene: {scene_summary}.\n" f"Controls: genre={genre_label}; tone={tone_label}; template={template_label}.\n" "Use these real objects as inspiration: " + ", ".join(object_names) + ".\n" "Rules: JSON only. No markdown. No tiles. No coordinates. No full npcs/items/landmarks arrays. " "No quest object. No asset filenames. No nested scene object. No instruction object. " "Keep it under 350 tokens." ) def _strip_markdown_fences(text): stripped = text.strip() fence_match = re.fullmatch(r"```(?:json)?\s*(.*?)\s*```", stripped, flags=re.IGNORECASE | re.DOTALL) if fence_match: return fence_match.group(1).strip() stripped = re.sub(r"^\s*```(?:json)?\s*", "", stripped, flags=re.IGNORECASE) stripped = re.sub(r"\s*```\s*$", "", stripped) return stripped.strip() def _strip_thinking_text(text): stripped = re.sub(r".*?", "", text, flags=re.IGNORECASE | re.DOTALL).strip() if stripped.lower().startswith(""): balanced = _first_balanced_json_object(stripped) if balanced: return balanced return stripped def _repair_prefilled_json_text(text): stripped = text.strip() if stripped.startswith("{,"): repaired = "{" + stripped[2:].lstrip() if _first_balanced_json_object(repaired): return repaired later = stripped.find('{"', 2) if later >= 0: return stripped[later:] return stripped[2:].lstrip() if not stripped.startswith("{"): return stripped for field in ('"title"', '"schema_version"'): index = stripped.find(field) if index > 0: return "{" + stripped[index:] return stripped def _balanced_json_object_from(text, start): if start < 0: return None depth = 0 in_string = False escape = False for index in range(start, len(text)): char = text[index] if in_string: if escape: escape = False elif char == "\\": escape = True elif char == '"': in_string = False continue if char == '"': in_string = True elif char == "{": depth += 1 elif char == "}": depth -= 1 if depth == 0: return text[start:index + 1] return None def _first_balanced_json_object(text): return _balanced_json_object_from(text, text.find("{")) def detect_prompt_echo(text): lowered = str(text or "").lower() echo_markers = [ "{system", "you are a json completion engine", "repair this pocketworld", "invalid_world_or_raw_output", '"role"', "assistant", "user", ] return any(marker in lowered for marker in echo_markers) def extract_json_like_segment(text): cleaned = _repair_prefilled_json_text(_strip_thinking_text(_strip_markdown_fences(str(text or "")))) starts = [index for index, char in enumerate(cleaned) if char == "{"] useful_fields = ('"title"', '"tiles"', '"player_start"', '"quest"') fallback_candidate = None for start in starts: candidate = _balanced_json_object_from(cleaned, start) if not candidate: continue if fallback_candidate is None: fallback_candidate = candidate if any(field in candidate for field in useful_fields): return candidate return fallback_candidate def _is_repair_prompt_echo(parsed): if not isinstance(parsed, dict): return False return "task" in parsed or "invalid_world_or_raw_output" in parsed def extract_json_from_text(raw_text, require_world=True): if not isinstance(raw_text, str) or not raw_text.strip(): raise ValueError("Model returned empty text; expected a JSON object.") original_preview = _preview_text(raw_text, 260) stripped = _repair_prefilled_json_text(_strip_thinking_text(_strip_markdown_fences(raw_text))) candidates = [] world_candidate = extract_json_like_segment(stripped) if world_candidate: candidates.append(world_candidate) candidates.append(stripped) balanced = _first_balanced_json_object(stripped) if balanced and balanced not in candidates: candidates.append(balanced) last_error = None for candidate in candidates: try: parsed = json.loads(candidate) except json.JSONDecodeError as exc: last_error = exc continue if not isinstance(parsed, dict): raise ValueError("Parsed JSON must be an object at the top level.") if _is_repair_prompt_echo(parsed): raise ValueError("Parsed JSON is a repair prompt echo, not a PocketWorld world JSON.") if require_world: required_world_fields = {"title", "tiles", "player_start", "quest"} if not required_world_fields.intersection(parsed.keys()): last_error = ValueError("Parsed JSON does not look like a PocketWorld world.") continue return parsed, _preview_text(candidate) preview = stripped[:220].replace("\n", " ") thinking_hint = " Model output contained thinking text and no balanced JSON object." if "" in raw_text.lower() and not balanced else "" raise ValueError( f"Could not parse a JSON object from model output.{thinking_hint} " f"Preview: {preview or original_preview}. Parser error: {last_error}" ) def extract_json_object(raw_text): parsed, _ = extract_json_from_text(raw_text) return parsed def _model_device(model): try: return next(model.parameters()).device except StopIteration: return "cpu" def load_text_generation_model(): if _truthy_env(AI_DISABLE_ENV): raise RuntimeError(f"AI generation is disabled by {AI_DISABLE_ENV}=1.") if _TEXT_MODEL_CACHE["tokenizer"] is not None and _TEXT_MODEL_CACHE["model"] is not None: return _TEXT_MODEL_CACHE["tokenizer"], _TEXT_MODEL_CACHE["model"], _TEXT_MODEL_CACHE["torch"] try: import torch from transformers import AutoModelForCausalLM, AutoTokenizer except Exception as exc: raise RuntimeError( "MiniCPM text generation dependencies are not available. Install requirements.txt or use sample worlds." ) from exc tokenizer = AutoTokenizer.from_pretrained(TEXT_WORLD_MODEL_ID, trust_remote_code=True) model_kwargs = {"trust_remote_code": True} if torch.cuda.is_available(): model_kwargs.update({"device_map": "auto", "torch_dtype": torch.float16}) elif getattr(torch.backends, "mps", None) and torch.backends.mps.is_available(): model_kwargs.update({"torch_dtype": torch.float16}) model = AutoModelForCausalLM.from_pretrained(TEXT_WORLD_MODEL_ID, **model_kwargs) if not torch.cuda.is_available(): if getattr(torch.backends, "mps", None) and torch.backends.mps.is_available(): model.to("mps") else: model.to("cpu") model.eval() _TEXT_MODEL_CACHE.update({"tokenizer": tokenizer, "model": model, "torch": torch}) return tokenizer, model, torch def _extract_generated_text(generation_output, tokenizer, prompt_text, prompt_token_count): if isinstance(generation_output, str): completion = generation_output.removeprefix(prompt_text).strip() return { "completion": completion, "full_output_preview": _preview_text(generation_output), "new_token_count": None, } if isinstance(generation_output, list) and generation_output: first = generation_output[0] if isinstance(first, dict): text = first.get("generated_text") or first.get("text") or "" completion = str(text).removeprefix(prompt_text).strip() return { "completion": completion, "full_output_preview": _preview_text(text), "new_token_count": None, } if isinstance(first, str): completion = first.removeprefix(prompt_text).strip() return { "completion": completion, "full_output_preview": _preview_text(first), "new_token_count": None, } sequences = getattr(generation_output, "sequences", generation_output) try: first_sequence = sequences[0] generated_ids = first_sequence[prompt_token_count:] decoded = tokenizer.decode(generated_ids, skip_special_tokens=True).strip() full_decoded = tokenizer.decode(first_sequence, skip_special_tokens=True).strip() completion = decoded or full_decoded.removeprefix(prompt_text).strip() return { "completion": completion, "full_output_preview": _preview_text(full_decoded), "new_token_count": int(generated_ids.shape[-1]) if hasattr(generated_ids, "shape") else len(generated_ids), } except Exception: text = str(generation_output) return { "completion": text.removeprefix(prompt_text).strip(), "full_output_preview": _preview_text(text), "new_token_count": None, } def _apply_chat_template_for_generation(tokenizer, prompt, prompt_mode): system_message = ( "You are a JSON completion engine. Output JSON only. " "Never output , reasoning, markdown, or explanations." ) if prompt_mode == "chat_template" and hasattr(tokenizer, "apply_chat_template"): messages = [ {"role": "system", "content": system_message}, {"role": "user", "content": prompt}, ] template_kwargs = { "tokenize": False, "add_generation_prompt": True, } try: return tokenizer.apply_chat_template( messages, **template_kwargs, enable_thinking=False, ) except TypeError: try: return tokenizer.apply_chat_template(messages, **template_kwargs) except Exception: pass except Exception: pass return ( "Return one valid JSON object and nothing else.\n" "No markdown. No explanation. First character must be { and last character must be }.\n\n" f"{prompt}\n" ) def _clean_prefilled_completion(prefill, completion): if not prefill: return completion.strip() stripped = completion.lstrip() if stripped.startswith(prefill): return stripped if stripped.startswith(","): return stripped.lstrip(",").lstrip() return (prefill + stripped).strip() def generate_model_text(prompt, max_new_tokens=1400, loaded_model=None, prompt_mode="chat_template", use_prefill=False): if _truthy_env(AI_FORCE_FAIL_ENV): raise RuntimeError(f"Forced AI generation failure from {AI_FORCE_FAIL_ENV}=1.") tokenizer, model, torch = loaded_model or load_text_generation_model() prompt_text = _apply_chat_template_for_generation(tokenizer, prompt, prompt_mode) prefill = "{" if use_prefill else "" prompt_text_for_model = prompt_text + prefill inputs = tokenizer(prompt_text_for_model, return_tensors="pt") device = _model_device(model) inputs = {key: value.to(device) for key, value in inputs.items()} input_token_count = int(inputs["input_ids"].shape[-1]) pad_token_id = tokenizer.pad_token_id if tokenizer.pad_token_id is not None else tokenizer.eos_token_id max_time = _float_env(AI_MAX_GENERATION_SECONDS_ENV, 120) with torch.no_grad(): generation_output = model.generate( **inputs, max_new_tokens=max_new_tokens, max_time=max_time, do_sample=False, repetition_penalty=1.05, pad_token_id=pad_token_id, eos_token_id=tokenizer.eos_token_id, ) extracted = _extract_generated_text( generation_output, tokenizer, prompt_text_for_model, input_token_count, ) completion = _clean_prefilled_completion(prefill, extracted["completion"]) result = { "text": completion, "input_token_count": input_token_count, "new_token_count": extracted.get("new_token_count"), "decoded_full_output_preview": extracted.get("full_output_preview", ""), "decoded_completion_preview": _preview_text(completion, 1500), "prompt_mode": prompt_mode, "json_prefill_used": bool(use_prefill), "prompt_echo_detected": detect_prompt_echo(completion), } if _truthy_env(AI_DEBUG_MODEL_OUTPUT_ENV): _debug_log( "debug model output", { "prompt_mode": prompt_mode, "input_token_count": result["input_token_count"], "new_token_count": result["new_token_count"], "completion_preview": result["decoded_completion_preview"], "prompt_echo_detected": result["prompt_echo_detected"], }, ) return result def _prepare_generated_world(raw_world, scene_parse, genre_label, catalog): genre_config = AI_GENRE_CONFIG.get(genre_label, AI_GENRE_CONFIG["Cozy Fantasy"]) world = copy.deepcopy(raw_world) world.setdefault("title", f"{scene_parse.get('source_scene', 'Scene')} PocketWorld") world.setdefault("intro", f"A tiny world generated from the {scene_parse.get('source_scene', 'source scene')}.") world.setdefault("genre", genre_config["genre"]) world.setdefault("theme", genre_config["theme"]) world["source"] = { "kind": "model_generated", "source_scene": scene_parse.get("source_scene"), "scene_type": scene_parse.get("scene_type"), "mood": scene_parse.get("mood"), "objects": _scene_source_objects(scene_parse), } return normalize_world_assets(world, catalog) def validate_generated_world_contract(world, world_size): errors = [] if not isinstance(world.get("title"), str) or not world["title"].strip(): errors.append("title must be a non-empty string") if not isinstance(world.get("intro"), str) or not world["intro"].strip(): errors.append("intro must be a non-empty string") if len(world.get("npcs", [])) < 2: errors.append("AI worlds must include at least 2 NPCs") if len(world.get("landmarks", [])) < 2: errors.append("AI worlds must include at least 2 landmarks") if len(world.get("items", [])) < 2: errors.append("AI worlds must include at least 2 items") if len(world.get("grounding", [])) < 4: errors.append("AI worlds must include at least 4 grounding entries") steps = world.get("quest_steps", []) step_types = [step.get("type") for step in steps if isinstance(step, dict)] for required_type in ("talk", "inspect", "collect", "unlock"): if required_type not in step_types: errors.append(f"AI quest_steps must include a {required_type} step") if world_size == "Large": tiles = world.get("tiles", []) width = len(tiles[0]) if tiles and isinstance(tiles[0], str) else 0 height = len(tiles) if width < 20 or height < 14: errors.append("Large AI worlds must be at least 20x14 tiles") if errors: raise ValueError("; ".join(errors)) return True def _validate_generated_world(raw_world, scene_parse, genre_label, world_size, catalog): world = _prepare_generated_world(raw_world, scene_parse, genre_label, catalog) validate_world(world) validate_generated_world_contract(world, world_size) return world def make_fake_ai_world(scene_parse, genre_label, tone_label, template_label, world_size, catalog): genre_config = AI_GENRE_CONFIG.get(genre_label, AI_GENRE_CONFIG["Cozy Fantasy"]) theme = genre_config["theme"] theme_data = catalog["themes"][theme] width = 24 if world_size == "Large" else 20 height = 16 if world_size == "Large" else 14 rows = [] for y in range(height): if y == 0 or y == height - 1: rows.append("W" * width) continue row = ["W"] + ["."] * (width - 2) + ["W"] if 2 <= y <= 5: row[7] = "W" if height - 7 <= y <= height - 4: row[width - 9] = "W" rows.append("".join(row)) gate_y = height - 3 gate_x = width - 2 gate_row = list(rows[gate_y]) gate_row[gate_x] = "G" rows[gate_y] = "".join(gate_row) objects = scene_parse.get("objects", []) names = [obj.get("name", f"object {index + 1}") for index, obj in enumerate(objects) if isinstance(obj, dict)] while len(names) < 6: names.append(["guide token", "archive card", "gate light", "road charm", "spare key", "beacon"][len(names)]) npc_keys = theme_data.get("npc_sprite_keys", ["npc_citizen", "npc_merchant"]) item_keys = theme_data.get("item_sprite_keys", ["key", "gem"]) landmark_keys = theme_data.get("landmark_asset_keys", ["tower", "gate"]) required_item_id = "generated_key" final_gate_id = "generated_gate" source_scene = scene_parse.get("source_scene", "Scene") return { "title": f"Generated Test World: {source_scene}", "intro": f"A fake AI-generated {tone_label.lower()} {genre_label.lower()} test world built from {source_scene}.", "genre": genre_config["genre"], "theme": theme, "tile_palette": theme_data["tile_palette"], "player_sprite_key": theme_data["player_sprite_key"], "tiles": rows, "player_start": [2, height - 3], "regions": [ { "id": "arrival_path", "name": f"{names[3].title()} Arrival Path", "source_object": names[3], "description": "The starting route that bends around grounded scene objects.", }, { "id": "guide_corner", "name": f"{names[0].title()} Guide Corner", "source_object": names[0], "description": "A small meeting area where the first clue is explained.", }, { "id": "final_threshold", "name": f"{names[2].title()} Threshold", "source_object": names[2], "description": "The final locked edge of the pocket world.", }, ], "grounding": [ {"source_object": names[0], "world_object": f"{names[0].title()} Guide Corner", "role": "NPC village", "asset_key": npc_keys[0]}, {"source_object": names[1], "world_object": f"{names[1].title()} Archive", "role": "inspectable landmark", "asset_key": landmark_keys[0]}, {"source_object": names[2], "world_object": f"{names[2].title()} Gate", "role": "final gate", "asset_key": theme_data["tile_palette"]["G"]}, {"source_object": names[4], "world_object": "Generated Key", "role": "required item", "asset_key": item_keys[0]}, {"source_object": names[5], "world_object": "Optional Spark", "role": "optional item", "asset_key": item_keys[1 % len(item_keys)]}, ], "npcs": [ { "id": "generated_guide", "name": f"{names[0].title()} Guide", "x": 3, "y": height - 4, "sprite_key": npc_keys[0], "role": "guide", "dialogue": f"This is a fake AI test path. Inspect the {names[1].title()} Archive, then find the Generated Key.", }, { "id": "generated_helper", "name": f"{names[3].title()} Helper", "x": 5, "y": 2, "sprite_key": npc_keys[1 % len(npc_keys)], "role": "lorekeeper", "dialogue": f"The {template_label} route is open, but the gate still wants a key.", }, ], "items": [ { "id": required_item_id, "name": "Generated Key", "x": width - 5, "y": height - 5, "sprite_key": item_keys[0], "description": f"A fake AI required item inspired by {names[4]}.", }, { "id": "optional_spark", "name": "Optional Spark", "x": 4, "y": 4, "sprite_key": item_keys[1 % len(item_keys)], "description": "An optional collectible proving multi-item rendering still works.", }, ], "landmarks": [ { "id": "generated_archive", "name": f"{names[1].title()} Archive", "x": width // 2, "y": height // 2, "sprite_key": landmark_keys[0], "source_object": names[1], "description": f"A fake AI landmark grounded in {names[1]}.", }, { "id": "generated_beacon", "name": f"{names[5].title()} Beacon", "x": width - 5, "y": 4, "sprite_key": landmark_keys[1 % len(landmark_keys)], "source_object": names[5], "description": f"A second landmark grounded in {names[5]}.", }, ], "quest": { "goal": "Follow the fake AI trace path, inspect the archive, collect the Generated Key, and unlock the gate.", "goal_id": final_gate_id, "required_item": required_item_id, "success_ending": "The fake AI world validates, renders, and proves the AI callback path is active.", }, "quest_steps": [ {"id": "talk_generated_guide", "type": "talk", "target": "generated_guide", "text": "Talk to the Generated Guide."}, {"id": "inspect_generated_archive", "type": "inspect", "target": "generated_archive", "text": "Inspect the generated archive."}, {"id": "collect_generated_key", "type": "collect", "target": required_item_id, "text": "Find the Generated Key."}, {"id": "unlock_generated_gate", "type": "unlock", "target": final_gate_id, "text": "Unlock the generated gate."}, ], } def _selected_controls(source_scene_name, genre_label, tone_label, template_label, world_size, display_theme, use_fake_ai): return { "source_scene": source_scene_name, "genre": genre_label, "tone": tone_label, "game_template": template_label, "world_size": world_size, "display_theme": display_theme, "use_fake_ai_world_generation": bool(use_fake_ai), } def _new_ai_trace(selected_controls, scene_parse): return { "source": "ai_generation", "selected_controls": selected_controls, "mock_scene_parse": scene_parse, "model_load": {"attempted": False, "success": False, "error": None}, "generation": { "attempted": False, "success": False, "raw_output": "", "error": None, "elapsed_seconds": None, "input_token_count": None, "new_token_count": None, "decoded_full_output_preview": "", "decoded_completion_preview": "", "prompt_echo_detected": False, }, "attempts": [], "json_parse": { "attempted": False, "success": False, "error": None, "extracted_json_preview": "", }, "validation": {"attempted": False, "success": False, "errors": []}, "repair": {"attempted": False, "success": False, "raw_output": "", "errors": []}, "fallback": {"used": False, "reason": None, "sample_world": None}, } def _fallback_sample_name_for_genre(genre_label): return { "Cozy Fantasy": "Moonwell Harbor", "Sci-Fi": "Blue Screen Station", "Haunted Mystery": "Archive Garden", "Tiny City": "Cableblock Crossing", }.get(genre_label, "Moonwell Harbor") def _render_world_response(world, display_theme, status_message, trace): return ( make_game_html(world, display_theme), world, make_grounding_panel(world, display_theme), make_world_score_panel(world, display_theme), status_message, trace, world, ) def _fallback_ai_response(genre_label, display_theme, reason, trace): fallback_name = _fallback_sample_name_for_genre(genre_label) reason_text = reason if isinstance(reason, str) else json.dumps(reason, ensure_ascii=True, default=str) _debug_log("fallback reason", {"fallback": fallback_name, "reason": reason}) fallback_world = get_world(fallback_name) trace["fallback"] = { "used": True, "reason": reason_text, "sample_world": fallback_name, } status = ( f"AI generation failed. Loaded fallback sample world: {fallback_name}.\n" f"Reason: {reason_text}" ) response = _render_world_response(fallback_world, display_theme, status, trace) _debug_log("render complete", {"world": fallback_world.get("title"), "fallback": True}) return response def _parse_model_world(raw_output, trace): _debug_log("JSON extraction start") trace["json_parse"]["attempted"] = True try: parsed_world, preview = extract_json_from_text(raw_output) trace["json_parse"].update( { "success": True, "error": None, "extracted_json_preview": preview, } ) _debug_log("JSON parse success", {"preview": preview}) return parsed_world except Exception as exc: trace["json_parse"].update( { "success": False, "error": str(exc), "extracted_json_preview": _preview_text(raw_output), } ) _debug_log("JSON parse failure", {"error": str(exc)}) _debug_log("traceback", traceback.format_exc()) raise def _validate_model_world(parsed_world, scene_parse, genre_label, world_size, catalog, trace): _debug_log("validation start") trace["validation"]["attempted"] = True try: world = _validate_generated_world(parsed_world, scene_parse, genre_label, world_size, catalog) trace["validation"].update({"success": True, "errors": []}) _debug_log("validation success", {"title": world.get("title")}) return world except Exception as exc: errors = [part.strip() for part in str(exc).split(";") if part.strip()] or [str(exc)] trace["validation"].update({"success": False, "errors": errors}) _debug_log("validation failure", {"errors": errors}) _debug_log("traceback", traceback.format_exc()) raise def _allowed_prompt_mode(value): return value if value in {"chat_template", "raw"} else "chat_template" def _record_generation_metadata(trace, model_result, elapsed): trace["generation"].update( { "attempted": True, "success": True, "raw_output": model_result["text"], "error": None, "elapsed_seconds": round(elapsed, 3), "input_token_count": model_result.get("input_token_count"), "new_token_count": model_result.get("new_token_count"), "decoded_full_output_preview": model_result.get("decoded_full_output_preview", ""), "decoded_completion_preview": model_result.get("decoded_completion_preview", ""), "prompt_echo_detected": model_result.get("prompt_echo_detected", False), } ) def _attempt_model_world_generation( attempt_name, prompt, prompt_mode, loaded_model, scene_parse, genre_label, world_size, catalog, trace, ): attempt = { "name": attempt_name, "prompt_mode": prompt_mode, "input_token_count": None, "new_token_count": None, "decoded_full_output_preview": "", "decoded_completion_preview": "", "raw_completion_preview": "", "prompt_echo_detected": False, "parse_success": False, "validation_success": False, "error": None, } use_prefill = _truthy_env(AI_USE_JSON_PREFILL_ENV) _debug_log("generation start", {"mode": "model", "attempt": attempt_name, "prompt_mode": prompt_mode, "json_prefill": use_prefill}) started_at = time.perf_counter() try: model_result = generate_model_text( prompt, max_new_tokens=1400, loaded_model=loaded_model, prompt_mode=prompt_mode, use_prefill=use_prefill, ) except Exception as exc: elapsed = time.perf_counter() - started_at attempt.update({"elapsed_seconds": round(elapsed, 3), "error": str(exc)}) trace["attempts"].append(attempt) trace["generation"].update( { "attempted": True, "success": False, "error": str(exc), "elapsed_seconds": round(elapsed, 3), } ) _debug_log("generation failure", {"attempt": attempt_name, "error": str(exc), "elapsed_seconds": round(elapsed, 3)}) _debug_log("traceback", traceback.format_exc()) raise elapsed = time.perf_counter() - started_at attempt.update( { "input_token_count": model_result.get("input_token_count"), "new_token_count": model_result.get("new_token_count"), "decoded_full_output_preview": model_result.get("decoded_full_output_preview", ""), "decoded_completion_preview": model_result.get("decoded_completion_preview", ""), "raw_completion_preview": model_result.get("decoded_completion_preview", ""), "prompt_echo_detected": model_result.get("prompt_echo_detected", False), "elapsed_seconds": round(elapsed, 3), } ) _record_generation_metadata(trace, model_result, elapsed) completion = model_result["text"] _debug_log("generation finished", {"attempt": attempt_name, "elapsed_seconds": round(elapsed, 3), "input_tokens": attempt["input_token_count"], "new_tokens": attempt["new_token_count"]}) _debug_log("completion first 1500 chars", completion[:1500]) if attempt["prompt_echo_detected"]: attempt["error"] = "Prompt echo detected in completion-only text." trace["attempts"].append(attempt) if _truthy_env(AI_DEBUG_MODEL_OUTPUT_ENV): _debug_log("debug parse error", attempt["error"]) raise ValueError(attempt["error"]) try: parsed_world = _parse_model_world(completion, trace) attempt["parse_success"] = True except Exception as exc: attempt["error"] = str(exc) trace["attempts"].append(attempt) if _truthy_env(AI_DEBUG_MODEL_OUTPUT_ENV): _debug_log("debug parse error", str(exc)) raise try: world = _validate_model_world(parsed_world, scene_parse, genre_label, world_size, catalog, trace) attempt["validation_success"] = True trace["attempts"].append(attempt) return world, parsed_world, completion except Exception as exc: attempt["error"] = str(exc) trace["attempts"].append(attempt) if _truthy_env(AI_DEBUG_MODEL_OUTPUT_ENV): _debug_log("debug validation errors", trace["validation"]["errors"]) raise def _looks_like_repairable_world_text(text): candidate = extract_json_like_segment(text) if not candidate: return False, None lowered = candidate.lower() if "invalid_world_or_raw_output" in lowered or '"task"' in lowered: return False, None useful = any(field in candidate for field in ('"title"', '"tiles"', '"player_start"', '"quest"')) return useful, candidate if useful else None def generate_ai_world_full_world_legacy( source_scene_name, genre_label, tone_label, template_label, world_size, use_fake_ai=False, display_theme=DEFAULT_THEME, ): catalog = load_asset_catalog() source_scene_name = source_scene_name if source_scene_name in AI_SOURCE_SCENES else "Chaotic Desk" genre_label = genre_label if genre_label in AI_GENRES else "Cozy Fantasy" tone_label = tone_label if tone_label in AI_TONES else "Whimsical" template_label = template_label if template_label in AI_GAME_TEMPLATES else "Quest Gate" world_size = world_size if world_size in AI_WORLD_SIZES else "Medium" display_theme = normalize_theme(display_theme) scene_parse = copy.deepcopy(MOCK_SCENE_PARSES[source_scene_name]) selected_controls = _selected_controls( source_scene_name, genre_label, tone_label, template_label, world_size, display_theme, use_fake_ai, ) trace = _new_ai_trace(selected_controls, scene_parse) trace["prompt_summary"] = _prompt_summary( scene_parse, genre_label, tone_label, template_label, world_size, catalog, ) _debug_log("Generate AI World clicked") _debug_log("fake AI mode", {"use_fake_ai": bool(use_fake_ai)}) _debug_log("selected controls", selected_controls) _debug_log("mock scene parse selected", scene_parse) loaded_model = None raw_output = "" if use_fake_ai: try: _debug_log("generation start", {"mode": "fake"}) started_at = time.perf_counter() fake_world = make_fake_ai_world( scene_parse, genre_label, tone_label, template_label, world_size, catalog, ) raw_output = json.dumps(fake_world, ensure_ascii=True, indent=2) elapsed = time.perf_counter() - started_at trace["generation"].update( { "attempted": True, "success": True, "raw_output": raw_output, "error": None, "elapsed_seconds": round(elapsed, 3), } ) _debug_log("generation finished", {"mode": "fake", "elapsed_seconds": round(elapsed, 3)}) _debug_log("raw output first 1000 chars", raw_output[:1000]) parsed_world = _parse_model_world(raw_output, trace) world = _validate_model_world(parsed_world, scene_parse, genre_label, world_size, catalog, trace) response = _render_world_response(world, display_theme, "Fake AI world generated and validated.", trace) _debug_log("render complete", {"world": world.get("title"), "fallback": False}) return response except Exception as exc: _debug_log("fake AI failure", {"error": str(exc)}) _debug_log("traceback", traceback.format_exc()) return _fallback_ai_response(genre_label, display_theme, str(exc), trace) if _truthy_env(AI_FORCE_FAIL_ENV): reason = f"Forced AI generation failure from {AI_FORCE_FAIL_ENV}=1." trace["generation"].update( { "attempted": True, "success": False, "raw_output": "", "error": reason, "elapsed_seconds": 0, } ) _debug_log("generation failure", {"error": reason, "forced": True}) return _fallback_ai_response(genre_label, display_theme, reason, trace) _debug_log("model load start", {"model": TEXT_WORLD_MODEL_ID}) trace["model_load"]["attempted"] = True try: loaded_model = load_text_generation_model() trace["model_load"].update({"success": True, "error": None}) _debug_log("model load success", {"model": TEXT_WORLD_MODEL_ID}) except Exception as exc: trace["model_load"].update({"success": False, "error": str(exc)}) _debug_log("model load failure", {"error": str(exc)}) _debug_log("traceback", traceback.format_exc()) return _fallback_ai_response(genre_label, display_theme, str(exc), trace) prompt = build_compact_world_generation_prompt( scene_parse, genre_label, tone_label, template_label, world_size, catalog, ) _debug_log("prompt built", {"builder": "compact", "chars": len(prompt)}) prompt_mode = _allowed_prompt_mode(os.getenv(AI_PROMPT_MODE_ENV, "chat_template")) attempt_plan = [(f"{prompt_mode}_no_prefill", prompt_mode)] if prompt_mode != "raw": attempt_plan.append(("raw_prompt_no_prefill", "raw")) last_error = None last_completion = "" last_parsed_world = None for attempt_name, attempt_prompt_mode in attempt_plan: try: world, last_parsed_world, last_completion = _attempt_model_world_generation( attempt_name, prompt, attempt_prompt_mode, loaded_model, scene_parse, genre_label, world_size, catalog, trace, ) response = _render_world_response(world, display_theme, "AI world generated and validated.", trace) _debug_log("render complete", {"world": world.get("title"), "fallback": False}) return response except Exception as exc: last_error = str(exc) last_completion = trace["generation"].get("raw_output", "") _debug_log("attempt failed", {"attempt": attempt_name, "error": last_error}) if "Prompt echo detected" in last_error and attempt_prompt_mode != "raw": _debug_log("retrying with raw prompt mode after prompt echo") continue if attempt_prompt_mode != "raw" and trace["json_parse"].get("success") is False: _debug_log("retrying with raw prompt mode after parse failure") continue repairable, repair_candidate = _looks_like_repairable_world_text(last_completion) if repairable and loaded_model is not None: trace["repair"]["attempted"] = True _debug_log("repair start") try: repair_prompt = build_repair_prompt( repair_candidate, trace["validation"]["errors"] or trace["json_parse"]["error"] or "unknown generation error", scene_parse, genre_label, tone_label, template_label, world_size, catalog, ) repair_result = generate_model_text( repair_prompt, max_new_tokens=1400, loaded_model=loaded_model, prompt_mode="raw", use_prefill=_truthy_env(AI_USE_JSON_PREFILL_ENV), ) repair_raw = repair_result["text"] trace["repair"]["raw_output"] = repair_raw _debug_log("repair raw completion first 1500 chars", repair_raw[:1500]) if detect_prompt_echo(repair_raw): raise ValueError("Repair attempt echoed prompt instead of returning completion-only JSON.") repaired_json, repair_preview = extract_json_from_text(repair_raw) trace["json_parse"]["extracted_json_preview"] = repair_preview repaired_world = _validate_model_world( repaired_json, scene_parse, genre_label, world_size, catalog, trace, ) trace["repair"].update({"success": True, "errors": []}) _debug_log("repair success", {"world": repaired_world.get("title")}) response = _render_world_response( repaired_world, display_theme, "AI world generated and validated.", trace, ) _debug_log("render complete", {"world": repaired_world.get("title"), "fallback": False}) return response except Exception as repair_exc: repair_errors = [part.strip() for part in str(repair_exc).split(";") if part.strip()] or [str(repair_exc)] trace["repair"].update({"success": False, "errors": repair_errors}) _debug_log("repair failure", {"errors": repair_errors}) _debug_log("traceback", traceback.format_exc()) last_error = str(repair_exc) else: trace["repair"].update( { "attempted": False, "success": False, "errors": ["No repairable partial PocketWorld JSON found in completion."], } ) prompt_echo = any(attempt.get("prompt_echo_detected") for attempt in trace.get("attempts", [])) if prompt_echo: fallback_reason = "Model echoed prompt instead of returning completion-only JSON." else: fallback_reason = last_error or trace["json_parse"]["error"] or trace["validation"]["errors"] or "AI generation failed." return _fallback_ai_response(genre_label, display_theme, fallback_reason, trace) def _new_design_patch_trace(selected_controls, scene_parse): return { "source": "ai_generation", "generation_mode": "design_patch", "selected_controls": selected_controls, "mock_scene_parse": scene_parse, "raw_model_output": "", "model_load": {"attempted": False, "success": False, "error": None}, "generation": { "attempted": False, "success": False, "error": None, "elapsed_seconds": None, "input_token_count": None, "new_token_count": None, "decoded_full_output_preview": "", "decoded_completion_preview": "", "prompt_echo_detected": False, }, "design_patch_parse": { "success": False, "error": None, "patch": None, }, "deterministic_builder": { "template_used": None, "success": False, "error": None, }, "world_validation": { "success": False, "errors": [], }, "fallback": {"used": False, "reason": None, "sample_world": None}, } def _fallback_design_patch_response(genre_label, display_theme, reason, trace): fallback_name = _fallback_sample_name_for_genre(genre_label) reason_text = reason if isinstance(reason, str) else json.dumps(reason, ensure_ascii=True, default=str) _debug_log("design patch fallback reason", {"fallback": fallback_name, "reason": reason_text}) fallback_world = get_world(fallback_name) trace["fallback"] = { "used": True, "reason": reason_text, "sample_world": fallback_name, } status = ( f"AI design generation failed. Loaded fallback sample world: {fallback_name}.\n" f"Reason: {reason_text}" ) response = _render_world_response(fallback_world, display_theme, status, trace) _debug_log("render complete", {"world": fallback_world.get("title"), "fallback": True}) return response def _validate_built_world_for_trace(world, trace): try: validate_world(world) trace["world_validation"].update({"success": True, "errors": []}) return True except Exception as exc: errors = [part.strip() for part in str(exc).split(";") if part.strip()] or [str(exc)] trace["world_validation"].update({"success": False, "errors": errors}) raise def generate_ai_world_design_patch( source_scene_name, genre_label, tone_label, template_label, world_size, use_fake_ai=False, display_theme=DEFAULT_THEME, ): catalog = load_asset_catalog() source_scene_name = source_scene_name if source_scene_name in AI_SOURCE_SCENES else "Chaotic Desk" genre_label = genre_label if genre_label in AI_GENRES else "Cozy Fantasy" tone_label = tone_label if tone_label in AI_TONES else "Whimsical" template_label = template_label if template_label in AI_GAME_TEMPLATES else "Quest Gate" world_size = world_size if world_size in AI_WORLD_SIZES else "Medium" display_theme = normalize_theme(display_theme) scene_parse = copy.deepcopy(MOCK_SCENE_PARSES[source_scene_name]) selected_controls = _selected_controls( source_scene_name, genre_label, tone_label, template_label, world_size, display_theme, use_fake_ai, ) selected_controls["ai_mode"] = "design_patch" trace = _new_design_patch_trace(selected_controls, scene_parse) _debug_log("Generate AI World clicked") _debug_log("AI mode", {"mode": "design_patch"}) _debug_log("fake AI mode", {"use_fake_ai": bool(use_fake_ai)}) _debug_log("selected controls", selected_controls) _debug_log("mock scene parse selected", scene_parse) if use_fake_ai: try: _debug_log("design patch generation start", {"mode": "fake"}) started_at = time.perf_counter() patch = make_fake_design_patch(scene_parse, genre_label, tone_label, template_label) raw_output = json.dumps(patch, ensure_ascii=True, indent=2) elapsed = time.perf_counter() - started_at trace["raw_model_output"] = raw_output trace["generation"].update( { "attempted": True, "success": True, "error": None, "elapsed_seconds": round(elapsed, 3), "decoded_completion_preview": _preview_text(raw_output, 1500), } ) trace["design_patch_parse"].update({"success": True, "error": None, "patch": patch}) world, template_used = build_world_from_design_patch( patch, scene_parse, genre_label, tone_label, template_label, world_size, catalog, ) trace["deterministic_builder"].update({"template_used": template_used, "success": True, "error": None}) _validate_built_world_for_trace(world, trace) response = _render_world_response( world, display_theme, "Fake AI design generated. Playable world built and validated.", trace, ) _debug_log("render complete", {"world": world.get("title"), "fallback": False, "template": template_used}) return response except Exception as exc: _debug_log("fake design patch failure", {"error": str(exc)}) _debug_log("traceback", traceback.format_exc()) return _fallback_design_patch_response(genre_label, display_theme, str(exc), trace) if _truthy_env(AI_FORCE_FAIL_ENV): reason = f"Forced AI generation failure from {AI_FORCE_FAIL_ENV}=1." trace["generation"].update({"attempted": True, "success": False, "error": reason, "elapsed_seconds": 0}) _debug_log("design patch generation failure", {"error": reason, "forced": True}) return _fallback_design_patch_response(genre_label, display_theme, reason, trace) _debug_log("model load start", {"model": TEXT_WORLD_MODEL_ID}) trace["model_load"]["attempted"] = True try: loaded_model = load_text_generation_model() trace["model_load"].update({"success": True, "error": None}) _debug_log("model load success", {"model": TEXT_WORLD_MODEL_ID}) except Exception as exc: trace["model_load"].update({"success": False, "error": str(exc)}) _debug_log("model load failure", {"error": str(exc)}) _debug_log("traceback", traceback.format_exc()) return _fallback_design_patch_response(genre_label, display_theme, str(exc), trace) prompt = build_design_patch_prompt(scene_parse, genre_label, tone_label, template_label) prompt_mode = _allowed_prompt_mode(os.getenv(AI_PROMPT_MODE_ENV, "raw")) _debug_log("design patch prompt built", {"chars": len(prompt), "prompt_mode": prompt_mode}) started_at = time.perf_counter() trace["generation"]["attempted"] = True try: model_result = generate_model_text( prompt, max_new_tokens=600, loaded_model=loaded_model, prompt_mode=prompt_mode, use_prefill=False, ) elapsed = time.perf_counter() - started_at raw_output = model_result["text"] trace["raw_model_output"] = raw_output trace["generation"].update( { "success": True, "error": None, "elapsed_seconds": round(elapsed, 3), "input_token_count": model_result.get("input_token_count"), "new_token_count": model_result.get("new_token_count"), "decoded_full_output_preview": model_result.get("decoded_full_output_preview", ""), "decoded_completion_preview": model_result.get("decoded_completion_preview", ""), "prompt_echo_detected": model_result.get("prompt_echo_detected", False), } ) _debug_log("design patch generation finished", {"elapsed_seconds": round(elapsed, 3), "new_tokens": model_result.get("new_token_count")}) _debug_log("design patch completion first 1500 chars", raw_output[:1500]) if model_result.get("prompt_echo_detected"): raise ValueError("Model echoed prompt instead of returning design patch JSON.") except Exception as exc: elapsed = time.perf_counter() - started_at trace["generation"].update({"success": False, "error": str(exc), "elapsed_seconds": round(elapsed, 3)}) _debug_log("design patch generation failure", {"error": str(exc), "elapsed_seconds": round(elapsed, 3)}) _debug_log("traceback", traceback.format_exc()) return _fallback_design_patch_response(genre_label, display_theme, str(exc), trace) try: parsed_patch, patch_preview = extract_json_from_text(raw_output, require_world=False) patch = validate_design_patch(parsed_patch, scene_parse) trace["design_patch_parse"].update({"success": True, "error": None, "patch": patch, "preview": patch_preview}) _debug_log("design patch parse success", {"title": patch.get("title")}) except Exception as exc: trace["design_patch_parse"].update({"success": False, "error": str(exc), "patch": None}) _debug_log("design patch parse failure", {"error": str(exc)}) _debug_log("traceback", traceback.format_exc()) return _fallback_design_patch_response(genre_label, display_theme, str(exc), trace) try: world, template_used = build_world_from_design_patch( patch, scene_parse, genre_label, tone_label, template_label, world_size, catalog, ) trace["deterministic_builder"].update({"template_used": template_used, "success": True, "error": None}) _validate_built_world_for_trace(world, trace) response = _render_world_response( world, display_theme, "AI design generated. Playable world built and validated.", trace, ) _debug_log("render complete", {"world": world.get("title"), "fallback": False, "template": template_used}) return response except Exception as exc: trace["deterministic_builder"].update({"success": False, "error": str(exc)}) _debug_log("deterministic builder or validation failure", {"error": str(exc)}) _debug_log("traceback", traceback.format_exc()) return _fallback_design_patch_response(genre_label, display_theme, str(exc), trace) def generate_ai_world( source_scene_name, genre_label, tone_label, template_label, world_size, use_fake_ai=False, display_theme=DEFAULT_THEME, ): ai_mode = os.getenv(AI_MODE_ENV, DEFAULT_AI_MODE).strip().lower() or DEFAULT_AI_MODE if ai_mode == "full_world_legacy" and not use_fake_ai: return generate_ai_world_full_world_legacy( source_scene_name, genre_label, tone_label, template_label, world_size, use_fake_ai, display_theme, ) return generate_ai_world_design_patch( source_scene_name, genre_label, tone_label, template_label, world_size, use_fake_ai, display_theme, ) def _safe_script_json(value): value_json = json.dumps(value, ensure_ascii=True, separators=(",", ":")) return value_json.replace("= 6 else "/file=" return file_route + path def _used_asset_defs(world, catalog): defs = {} for _, asset_key in _world_asset_refs(world): entry = _catalog_entry(catalog, asset_key) if entry and entry.get("path"): defs[asset_key] = { "key": asset_key, "category": entry.get("category"), "url": _asset_url(entry["path"]), "path": entry["path"], } return defs def make_game_html(world, display_theme=DEFAULT_THEME): catalog = load_asset_catalog() world = normalize_world_assets(world, catalog) validate_world(world) theme_mode = normalize_theme(display_theme).lower() used_assets = _used_asset_defs(world, catalog) world_json_for_js = json.dumps(_safe_script_json(world), ensure_ascii=True) assets_json = _safe_script_json(used_assets) map_width = len(world["tiles"][0]) map_height = len(world["tiles"]) viewport_width = min(map_width * 44, 704) viewport_height = min(map_height * 44, 512) camera_zoom = 1.12 if map_width > 16 or map_height > 12 else 1 iframe_height = max(720, min(860, viewport_height + 240)) inner_html = """

WASD/arrow keys to move. Press E to interact.

""" inner_html = ( inner_html .replace("__THEME_MODE__", theme_mode) .replace("__WORLD_JSON__", world_json_for_js) .replace("__ASSETS_JSON__", assets_json) .replace("__MAP_WIDTH__", str(map_width)) .replace("__MAP_HEIGHT__", str(map_height)) .replace("__VIEWPORT_WIDTH__", str(viewport_width)) .replace("__VIEWPORT_HEIGHT__", str(viewport_height)) .replace("__CAMERA_ZOOM__", str(camera_zoom)) ) return f""" """ def make_grounding_panel(world, display_theme=DEFAULT_THEME): world = normalize_world_assets(world) theme_class = _theme_class(display_theme) rows = [] for entry in world.get("grounding", []): source = html.escape(str(entry.get("source_object", "Unknown object"))) target = html.escape(str(entry.get("world_object", "PocketWorld object"))) role = html.escape(str(entry.get("role", "world element"))) rows.append(f"{source}{target}{role}") if not rows: rows.append("No grounding entries yet.") return f"""

Grounding Panel

{''.join(rows)}
Original objectPocketWorld objectRole
""" def make_world_score_panel(world, display_theme=DEFAULT_THEME): world = normalize_world_assets(world) score = compute_world_score(world) theme_class = _theme_class(display_theme) grounding = score["grounding"] asset_validity = score["asset_validity"] warnings = score.get("warnings", []) warning_html = "" if warnings: warning_items = "".join(f"
  • {html.escape(warning)}
    • " for warning in warnings[:6]) warning_html = f"
      {warning_items}
    " return f"""

    World Score

    Theme{html.escape(str(score['theme']).replace('_', ' '))}
    Grounding{grounding['used']} / {grounding['total']} objects
    Playability{'pass' if score['playability'] else 'fail'}
    Quest{'complete' if score['quest_complete'] else 'incomplete'}
    Assets{asset_validity['valid']} / {asset_validity['total']} valid
    {warning_html}
    """ def get_world(sample_name): if sample_name not in SAMPLE_WORLD_LOOKUP: raise gr.Error(f"Unknown sample world: {sample_name}") world = normalize_world_assets(SAMPLE_WORLD_LOOKUP[sample_name]) try: validate_world(world) except ValueError as exc: raise gr.Error(f"Invalid world JSON: {exc}") from exc return world def load_world(sample_name, display_theme=DEFAULT_THEME): world = get_world(sample_name) trace = { "source": "sample_world", "sample_world": sample_name, "validation_result": "pass", "fallback": {"used": False, "reason": None, "sample_world": None}, } status = f"Loaded sample world: {sample_name}" return _render_world_response(world, display_theme, status, trace) def rerender_current_world(current_world, display_theme=DEFAULT_THEME, status_message="", trace=None): world = normalize_world_assets(current_world or INITIAL_WORLD) validate_world(world) status = status_message or f"Re-rendered current world: {world.get('title', 'PocketWorld')}" current_trace = trace or { "source": "current_world", "validation_result": "pass", "fallback": {"used": False, "reason": None, "sample_world": None}, } return _render_world_response(world, display_theme, status, current_trace) def make_catalog_contract(): catalog = load_asset_catalog() return { "model_rule": "Choose only asset keys from asset_catalog.json. Never invent filenames.", "default_ai_mode": DEFAULT_AI_MODE, "design_patch_schema": DESIGN_PATCH_SCHEMA, "world_schema": WORLD_SCHEMA, "asset_catalog": catalog, } def _gradio_major_version(): version = getattr(gr, "__version__", "0") try: return int(str(version).split(".", 1)[0]) except (TypeError, ValueError): return 0 def _blocks_kwargs(): if _gradio_major_version() >= 6: return {} return {"css": CUSTOM_CSS} def _launch_kwargs(): kwargs = {"allowed_paths": [str(ASSETS_DIR)]} if _gradio_major_version() >= 6: kwargs["css"] = CUSTOM_CSS return kwargs INITIAL_WORLD_NAME = SAMPLE_WORLDS[0]["title"] INITIAL_WORLD = get_world(INITIAL_WORLD_NAME) INITIAL_STATUS = f"Loaded sample world: {INITIAL_WORLD_NAME}" INITIAL_TRACE = { "source": "sample_world", "sample_world": INITIAL_WORLD_NAME, "validation_result": "pass", "fallback": {"used": False, "reason": None, "sample_world": None}, } with gr.Blocks(**_blocks_kwargs()) as demo: gr.Markdown("# PocketWorld Studio") gr.Markdown( "WIP: Phase 4A adds text-to-world generation from mock scene parses. " "No image upload, vision model, or chatbot flow yet.", elem_classes="pw-note", ) gr.Markdown("Controls: WASD/arrow keys to move, E to interact.") with gr.Group(elem_classes="pw-ai-box"): gr.Markdown("## Generate AI World") gr.Markdown( "MiniCPM is loaded only when you click Generate AI World. " "The model outputs world JSON only; Phaser gameplay still runs entirely in the browser.", elem_classes="pw-ai-muted", ) with gr.Row(): source_scene_dropdown = gr.Dropdown( choices=AI_SOURCE_SCENES, value="Chaotic Desk", label="Source scene", ) ai_genre_dropdown = gr.Dropdown( choices=AI_GENRES, value="Cozy Fantasy", label="Genre", ) ai_tone_dropdown = gr.Dropdown( choices=AI_TONES, value="Whimsical", label="Tone", ) with gr.Row(): ai_template_dropdown = gr.Dropdown( choices=AI_GAME_TEMPLATES, value="Quest Gate", label="Game template", ) ai_world_size_dropdown = gr.Dropdown( choices=AI_WORLD_SIZES, value="Large", label="World size", ) use_fake_ai_checkbox = gr.Checkbox( value=False, label="Use fake AI world generation", info="Smoke-test the AI path without loading MiniCPM.", ) generate_button = gr.Button("Generate AI World", variant="primary") ai_status_output = gr.Textbox( value=INITIAL_STATUS, label="Generation status", lines=5, interactive=False, ) with gr.Row(): sample_dropdown = gr.Dropdown( choices=list(SAMPLE_WORLD_LOOKUP.keys()), value=INITIAL_WORLD_NAME, label="Sample world", ) theme_dropdown = gr.Dropdown( choices=THEME_OPTIONS, value=DEFAULT_THEME, label="Display theme", info="Affects the game iframe and PocketWorld panels.", ) load_button = gr.Button("Load World", variant="primary") with gr.Row(equal_height=False, elem_classes="pw-game-shell"): with gr.Column(scale=3, min_width=320): game_output = gr.HTML(value=make_game_html(INITIAL_WORLD, DEFAULT_THEME), label="Game") with gr.Column(scale=2, min_width=280): grounding_output = gr.HTML( value=make_grounding_panel(INITIAL_WORLD, DEFAULT_THEME), label="Grounding Panel", ) score_output = gr.HTML( value=make_world_score_panel(INITIAL_WORLD, DEFAULT_THEME), label="World Score", ) with gr.Accordion("Currently loaded world JSON", open=False): world_json_output = gr.JSON(value=INITIAL_WORLD, label=None) with gr.Accordion("Generation Trace", open=False): generation_trace_output = gr.JSON(value=INITIAL_TRACE, label=None) with gr.Accordion("Asset catalog contract for future image/LLM models", open=False): gr.JSON(value=make_catalog_contract(), label=None) current_world_state = gr.State(INITIAL_WORLD) load_button.click( fn=load_world, inputs=[sample_dropdown, theme_dropdown], outputs=[ game_output, world_json_output, grounding_output, score_output, ai_status_output, generation_trace_output, current_world_state, ], ) theme_dropdown.change( fn=rerender_current_world, inputs=[ current_world_state, theme_dropdown, ai_status_output, generation_trace_output, ], outputs=[ game_output, world_json_output, grounding_output, score_output, ai_status_output, generation_trace_output, current_world_state, ], ) generate_button.click( fn=generate_ai_world, inputs=[ source_scene_dropdown, ai_genre_dropdown, ai_tone_dropdown, ai_template_dropdown, ai_world_size_dropdown, use_fake_ai_checkbox, theme_dropdown, ], outputs=[ game_output, world_json_output, grounding_output, score_output, ai_status_output, generation_trace_output, current_world_state, ], ) def run_phase4a_self_check(): _debug_log("phase4a self-check start") fake_outputs = generate_ai_world( "Chaotic Desk", "Cozy Fantasy", "Whimsical", "Quest Gate", "Large", True, "Auto", ) fake_world = fake_outputs[1] fake_trace = fake_outputs[5] assert fake_world["title"] == "Generated Test World: Chaotic Desk" assert fake_trace["source"] == "ai_generation" assert fake_trace["generation_mode"] == "design_patch" assert fake_trace["fallback"]["used"] is False assert fake_trace["model_load"]["attempted"] is False assert fake_trace["design_patch_parse"]["success"] is True assert fake_trace["deterministic_builder"]["success"] is True assert fake_trace["world_validation"]["success"] is True validate_world(fake_world) extraction_world = copy.deepcopy(fake_world) extraction_world["title"] = "Plain JSON World" extraction_json = json.dumps(extraction_world, ensure_ascii=True) plain_json, plain_preview = extract_json_from_text(extraction_json) assert plain_json["title"] == "Plain JSON World" assert "Plain JSON World" in plain_preview fenced_json, fenced_preview = extract_json_from_text(f"```json\n{extraction_json}\n```") assert fenced_json["title"] == "Plain JSON World" assert "Plain JSON World" in fenced_preview thinking_json, thinking_preview = extract_json_from_text(f"notes that should be ignored\n{extraction_json}") assert thinking_json["title"] == "Plain JSON World" assert "Plain JSON World" in thinking_preview prefilled_json, prefilled_preview = extract_json_from_text("{," + extraction_json[1:]) assert prefilled_json["title"] == "Plain JSON World" assert "Plain JSON World" in prefilled_preview try: extract_json_from_text('{"task":"Repair this PocketWorld world JSON so it validates.","invalid_world_or_raw_output":"{system"}') except ValueError as exc: assert "repair prompt echo" in str(exc) else: raise AssertionError("repair prompt echo unexpectedly parsed") try: extract_json_from_text("{bad json") except ValueError as exc: assert "Could not parse" in str(exc) else: raise AssertionError("invalid JSON unexpectedly parsed") previous_force = os.environ.get(AI_FORCE_FAIL_ENV) os.environ[AI_FORCE_FAIL_ENV] = "1" try: fallback_outputs = generate_ai_world( "Chaotic Desk", "Cozy Fantasy", "Whimsical", "Quest Gate", "Large", False, "Auto", ) finally: if previous_force is None: os.environ.pop(AI_FORCE_FAIL_ENV, None) else: os.environ[AI_FORCE_FAIL_ENV] = previous_force fallback_trace = fallback_outputs[5] assert fallback_trace["source"] == "ai_generation" assert fallback_trace["generation_mode"] == "design_patch" assert fallback_trace["fallback"]["used"] is True assert fallback_trace["fallback"]["sample_world"] == "Moonwell Harbor" assert "AI design generation failed. Loaded fallback sample world: Moonwell Harbor." in fallback_outputs[4] _debug_log("phase4a self-check passed") print("Phase 4A self-check passed.", flush=True) if __name__ == "__main__": if _truthy_env(AI_SELF_CHECK_ENV): run_phase4a_self_check() else: demo.launch(**_launch_kwargs())