backend/tools/xr_tools.py

import logging

async def xr_anchor_tool(action: str, anchor_id: str = "", *args, **kwargs):
    """
    Control or query XR spatial anchors.
    Action: 'save', 'get', 'clear'
    """
    from backend.ws.agent_ws import ws_manager
    await ws_manager.broadcast({
        "event": "xr:anchor",
        "payload": {"action": action, "anchor_id": anchor_id}
    })
    return f"Broadcasted XR anchor command: {action} on {anchor_id}"

async def gesture_context_tool(gesture: str, *args, **kwargs):
    """
    Fake/inject an XR gesture for testing or context (e.g. 'pinch', 'swipe').
    """
    from backend.ws.agent_ws import ws_manager
    await ws_manager.broadcast({
        "event": f"gesture-{gesture}",
        "payload": {"source": "jarvis"}
    })
    return f"Injected simulated gesture: {gesture}"

async def ar_control_tool(action: str, payload: str = "", *args, **kwargs):
    """
    Control AR states globally (e.g. toggle AR, spawn model, enter vr, clear scene).
    """
    from backend.ws.agent_ws import ws_manager
    await ws_manager.broadcast({
        "event": "ar:control",
        "payload": {"action": action, "data": payload}
    })
    return f"Dispatched AR control command: {action} with payload {payload}"

class StableFast3DNotProvisionedError(Exception):
    pass

async def is_exe_sidecar_reachable() -> bool:
    """Mock check if EXE sidecar is reachable."""
    return True

async def get_inference_target() -> str:
    if await is_exe_sidecar_reachable():
        return "exe_local_gpu"
    else:
        return "hf_space_cpu_fallback"

async def find_known_image(description: str) -> str | None:
    """Tier 1: Check SQLite then ChromaDB for a known prior 3D generation."""
    # Fast path: SQLite lookup
    from backend.memory import database
    row = await database.lookup_model_description(description)
    if row:
        return row["model_path"]
        
    # Semantic match: ChromaDB lookup
    from backend.memory.episodic_memory import EpisodicMemory
    em = EpisodicMemory("storage/chroma_db")
    similar = await em.query_similar_model(description, threshold=0.85)
    if similar:
        return similar["model_path"]
    return None

async def search_web_for_image(description: str) -> str | None:
    """Tier 2: Search the web (Wikimedia Commons) for a matching image."""
    import logging
    import urllib.parse
    import urllib.request
    import json
    import tempfile
    import asyncio
    
    logging.info(f"Tier 2: Searching web for image matching: {description}")
    def _search():
        # Using Wikipedia/Wikimedia API to find an open-source image
        url = f"https://en.wikipedia.org/w/api.php?action=query&prop=pageimages&format=json&piprop=original&titles={urllib.parse.quote(description)}"
        req = urllib.request.Request(url, headers={'User-Agent': 'JARVIS_OMEGA/1.0'})
        try:
            with urllib.request.urlopen(req, timeout=5) as resp:
                data = json.loads(resp.read().decode('utf-8'))
                pages = data.get("query", {}).get("pages", {})
                for page_id, page_data in pages.items():
                    if "original" in page_data:
                        img_url = page_data["original"]["source"]
                        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
                        urllib.request.urlretrieve(img_url, tmp.name)
                        return tmp.name
        except Exception as e:
            logging.error(f"Web image search failed: {e}")
        return None
        
    return await asyncio.to_thread(_search)

async def pollinations_generate_image(description: str) -> str:
    """Tier 3: Generating AI image as ultimate fallback."""
    import logging
    import urllib.parse
    import urllib.request
    import tempfile
    import asyncio
    
    logging.info(f"Tier 3: Generating AI image for: {description}")
    
    def _generate():
        # Fallback AI image generator (using free Pollinations API as proxy for Gemini output generation)
        # Note: In production this would use Gemini/Imagen via Vertex AI.
        prompt = urllib.parse.quote(f"A high quality realistic 3D asset of {description}, isolated on white background, studio lighting")
        url = f"https://image.pollinations.ai/prompt/{prompt}?width=1024&height=1024&nologo=true"
        try:
            tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
            urllib.request.urlretrieve(url, tmp.name)
            return tmp.name
        except Exception as e:
            logging.error(f"AI image generation failed: {e}")
            raise e
            
    return await asyncio.to_thread(_generate)

async def log_model_generation(description: str, image_tier: str, model_path: str, persona: str, inference_target: str):
    from backend.db import mongodb
    from backend.memory import database
    from backend.memory.episodic_memory import EpisodicMemory
    em = EpisodicMemory("storage/chroma_db")
    import asyncio
    await asyncio.gather(
        mongodb.log_model_generation_mongo(description, image_tier, model_path, persona, inference_target),
        em.embed_model_description(description, model_path, persona),
        database.insert_model_generation(description, image_tier, model_path, persona, inference_target)
    )

async def push_model_to_ar_scene(model_path: str, description: str):
    try:
        from phone import ws_scene_bus
        patch = {
            "type": "patch",
            "patch": {
                "action": "spawn-generated-glb",
                "model_url": f"/static/models3d/{model_path.split('/')[-1]}" if '/' in model_path else f"/static/models3d/{model_path}",
                "model_name": description
            }
        }
        # S4: broadcast_to_clients is sync — awaiting it raised TypeError and the
        # spawn broadcast never reached the AR clients. Use the async variant.
        await ws_scene_bus.broadcast_to_clients_async(patch)
    except Exception as e:
        logging.error(f"Failed to push model to AR scene bus: {e}")

async def spawn_3d_model(description: str, persona: str, use_cloud_fallback: bool = False) -> str:
    from backend.services.connectors import stable_fast_3d_local
    health = await stable_fast_3d_local.health_check()
    if not health["ready"]:
        raise StableFast3DNotProvisionedError("Stable Fast 3D model is not installed. Run scripts/provision_stable_fast_3d.sh first.")

    image_source_tier = "known"
    image = await find_known_image(description)

    if not image:
        image_source_tier = "web_search"
        image = await search_web_for_image(description)

    if not image:
        image_source_tier = "ai_generated"
        image = await pollinations_generate_image(description)

    model_path = await stable_fast_3d_local.image_to_3d_local(image, use_cloud_fallback)
    
    if model_path == "INTERACTIVE_PROMPT_REQUIRED":
        return "STATUS: LOCAL_MODEL_MISSING. Please ask the user: 'The local SF3D engine is not installed. Should I initiate the 10GB installation script, or route this through the Cloud API for now?' If they choose Cloud API, call this tool again with use_cloud_fallback=True."
        
    await push_model_to_ar_scene(model_path, description)
    
    # S4: determine_inference_target is synchronous — awaiting it raised TypeError
    # after every successful spawn, killing the Mongo/Chroma/SQLite logging below.
    target = stable_fast_3d_local.determine_inference_target()
    await log_model_generation(description, image_source_tier, model_path, persona, target)
    
    return model_path