backend/routes/mobile_bridge_routes.py

from __future__ import annotations

"""
backend/routes/mobile_bridge_routes.py — S4 Guardian ⇄ cloud REST compatibility.

The JARVIS Mobile Guardian APK (phone/jarvis-mobile-guardian) speaks a fixed
`/api/*` REST contract (see network/ApiClient.kt). On the local PC that contract
is served by phone/local_server.py on :7474; on the cloud it was only partially
served — xr_router@/api covers ar_task/ar_scene/ar_game_state/ar_game_builder,
but eight endpoints (pair_confirm, status, caps, link_info, command,
phone_observe, jarvis/mobile_event, max_autonomy/task) had no cloud handler, so
Guardian's status card, capability grid, AR-URL fetch, command box, telemetry
sync and autonomy tasks all 404'd against jarvis-cloud.hf.space.

This router fills exactly those gaps, mounted at /api behind verify_token
(Guardian sends `Authorization: Bearer <JARVIS_CLOUD_TOKEN>` on every call). It
delegates to the real cross-device coordinator (modules/max_autonomy.py — the
module explicitly built as "the single backend surface for APK, PC AR, browser
panels and Guardian heartbeats") and the shared AES-GCM helpers (phone/crypto.py)
rather than re-implementing behaviour, so cloud and PC stay in lock-step.

Encrypted /api/command: Guardian derives its AES key as SHA-256(passphrase)
(CryptoBox.deriveKey). phone/crypto.LinkConfig.key_bytes() derives the same way,
so a cloud pairing whose passphrase is the JARVIS_CLOUD_TOKEN lets the Space
decrypt and route the command through the safety-gated autonomy engine.
"""

import base64
import hashlib
import logging
import os
from backend.version import VERSION as _VERSION
import secrets
import time
from typing import Any

from fastapi import APIRouter, Request, Response

router = APIRouter()
log = logging.getLogger(__name__)

_BOOT_TS = time.time()


# --------------------------------------------------------------------------- #
# helpers
# --------------------------------------------------------------------------- #
def _master_token() -> str:
    try:
        from backend.dependencies.auth import get_or_create_master_token
        return get_or_create_master_token()
    except Exception:
        return os.environ.get("JARVIS_CLOUD_TOKEN", "").strip()


def _uptime_str() -> str:
    secs = int(time.time() - _BOOT_TS)
    h, rem = divmod(secs, 3600)
    m, s = divmod(rem, 60)
    if h:
        return f"{h}h {m}m"
    if m:
        return f"{m}m {s}s"
    return f"{s}s"


def _cpu_ram() -> tuple[float | None, float | None]:
    try:
        import psutil  # type: ignore
        return (
            float(psutil.cpu_percent(interval=None)),
            float(psutil.virtual_memory().percent),
        )
    except Exception:
        return (None, None)


def _base_url(request: Request) -> str:
    # Honour reverse-proxy host so the AR URL points at the public Space, not the
    # internal 127.0.0.1:7860 uvicorn bind.
    host = request.headers.get("x-forwarded-host") or request.headers.get("host")
    if host:
        proto = request.headers.get("x-forwarded-proto") or request.url.scheme
        return f"{proto}://{host}".rstrip("/")
    return str(request.base_url).rstrip("/")


async def _json_body(request: Request) -> dict[str, Any]:
    try:
        data = await request.json()
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _candidate_keys() -> list[bytes]:
    """AES keys to try when decrypting a Guardian command envelope, best-first."""
    keys: list[bytes] = []
    token = _master_token()
    if token:
        keys.append(hashlib.sha256(token.encode("utf-8")).digest())
    try:
        from phone.crypto import load_or_create_link
        keys.append(load_or_create_link().key_bytes())
    except Exception as _exc:
        log.debug("link key unavailable: %s", _exc)
    return keys


# --------------------------------------------------------------------------- #
# pairing / status / capabilities / link info
# --------------------------------------------------------------------------- #
@router.post("/pair_confirm")
async def pair_confirm(request: Request) -> dict[str, Any]:
    body = await _json_body(request)
    device_name = str(body.get("device_name") or "android").strip()[:64]
    passphrase = str(body.get("passphrase") or "").strip()
    # The Bearer token already authenticated the request; pairing simply records
    # the device. If the passphrase matches the cloud token, encrypted commands
    # will also decrypt cleanly (see /command).
    token = _master_token()
    keyed = bool(token) and secrets.compare_digest(passphrase, token)
    try:
        from phone.pairing import mark_paired
        mark_paired(device_name=device_name or "android")
    except Exception as _exc:
        log.debug("mark_paired skipped on cloud: %s", _exc)
    return {"ok": True, "paired": True, "keyed": keyed, "device_name": device_name}


@router.get("/status")
async def status() -> dict[str, Any]:
    cpu, ram = _cpu_ram()
    location = "cloud" if os.environ.get("SPACE_ID") else "pc"
    return {
        "ok": True,
        "status": "online",
        "version": _VERSION,
        "uptime": _uptime_str(),
        "mode": location,
        "activated": True,
        "assistant_name": "JARVIS",
        "cpu": cpu,
        "ram": ram,
    }


@router.get("/caps")
async def caps() -> dict[str, Any]:
    """Flatten the capability catalogue to the {capabilities:[str]} shape the APK
    deserializes (CapabilityList)."""
    labels: list[str] = []
    try:
        from phone.capabilities import capabilities_payload
        for category in capabilities_payload().get("categories", []):
            for item in category.get("items", []):
                label = str(item.get("label") or item.get("cmd") or "").strip()
                if label:
                    labels.append(label)
    except Exception as _exc:
        log.debug("capabilities unavailable: %s", _exc)
    return {"ok": True, "capabilities": labels}


@router.get("/link_status")
async def link_status() -> dict[str, Any]:
    """The truth about the phone -> cloud -> PC path, in one call.

    Guardian used to build its status cards from two unrelated signals: a /health
    ping (proves the Space is up) and a stored boolean. Neither said anything
    about whether a command could actually REACH the desktop, so the app happily
    rendered "PC CONTROL: CLOUD online" next to "NO ACTIVE LINK" while the phone
    could not send a single command. A status light that does not test the path
    it claims to describe is worse than no light at all.

    Reaching this endpoint at all proves the cloud leg. `pc_connected` is the
    number of desktops holding a live, registered WebSocket — the exact list
    send_to_pcs() delivers to — so it is the same fact the command path uses,
    not a parallel guess. `pc_unlocked` comes from the PC's own periodic report.
    """
    pc_count = 0
    status: dict[str, Any] = {}
    age: float | None = None
    queued = 0
    try:
        from backend.ws.agent_ws import ws_manager as _ws
        pc_count = len(_ws.pc_connections)
        status = dict(_ws.last_pc_status or {})
        queued = len(_ws.pending_commands)
        if _ws.last_pc_status_at:
            age = round(time.time() - _ws.last_pc_status_at, 1)
    except Exception as exc:
        log.warning("link_status unavailable: %s", exc)

    unlocked = bool(status.get("unlocked"))
    if pc_count <= 0:
        state, detail = "offline", "Your PC is not connected."
    elif not unlocked:
        state, detail = "locked", "Connected, but locked. Send: unlock pc <password>"
    else:
        state, detail = "ready", "Connected and ready for commands."

    return {
        "ok": True,
        "cloud": True,            # answering at all proves this leg
        "pc_connected": pc_count,
        "pc_unlocked": unlocked,
        "state": state,           # offline | locked | ready
        "detail": detail,
        "queued_commands": queued,
        "pc_status_age_seconds": age,
        "pc_status": status,
    }


@router.get("/link_info")
async def link_info(request: Request) -> dict[str, Any]:
    base = _base_url(request)
    ws_base = base.replace("https://", "wss://").replace("http://", "ws://")
    ar_url = f"{base}/webar/"
    return {
        "ok": True,
        "assistant": "JARVIS",
        "host": base,
        "pwa": ar_url,
        "pwa_https": ar_url,
        "ar": ar_url,
        "ar_https": ar_url,
        "ar_engine": "8thwall",
        "ar_runtime": "webxr",
        "scene_ws": f"{ws_base}/scene/ws",
    }


# --------------------------------------------------------------------------- #
# command (encrypted) — routed through the safety-gated autonomy engine
# --------------------------------------------------------------------------- #
@router.post("/command")
async def command(request: Request) -> dict[str, Any]:
    envelope = await _json_body(request)
    if not (envelope.get("nonce") and envelope.get("ciphertext")):
        return {"ok": False, "error": "expected encrypted envelope {nonce, ciphertext}"}

    plaintext: dict[str, Any] | None = None
    try:
        from phone.crypto import decrypt_json
        for key in _candidate_keys():
            try:
                plaintext = decrypt_json(envelope, key)
                break
            except Exception:
                continue
    except Exception as _exc:
        log.debug("decrypt_json import failed: %s", _exc)

    if not isinstance(plaintext, dict):
        # Wrong pairing passphrase for the cloud: pair with the cloud access key
        # so the Space can derive the same AES key.
        return {"ok": False, "error": "decrypt failed — pair using your cloud access key"}

    cmd = str(plaintext.get("cmd") or "").strip()
    if not cmd:
        return {"ok": False, "error": "missing cmd"}
    device_name = str(plaintext.get("device_name") or "").strip()[:80]

    # RELAY TO THE PC FIRST, if one is connected.
    #
    # The cloud container has no desktop of its own, which is why this used to go
    # straight to record-only mode. But it is not the only machine involved:
    # pc_relay_client.py runs on the user's PC, holds an outbound WebSocket to
    # this Space, and executes `system:execute` for real. Everything needed for
    # phone -> cloud -> PC control existed except this hop — the Space simply
    # never forwarded, so commands sent from anywhere in the world were logged
    # and dropped while the desktop sat connected and idle.
    #
    # Falls through to record-only when no PC is listening, which is the honest
    # answer when the machine is off rather than a failure.
    #
    # We wait for the PC's actual answer rather than reporting success on
    # hand-off. The desktop can refuse a command outright — an un-activated
    # session hard-blocks every one of them — and reporting "Sent to your PC."
    # in that case tells the user their command ran when it did not.
    try:
        from backend.ws.agent_ws import ws_manager as _ws_manager
        outcome = await _ws_manager.execute_on_pcs(cmd, device_name=device_name)
    except Exception as exc:
        log.warning("PC relay unavailable: %s", exc)
        outcome = {"delivered": 0, "ok": False, "spoken": [], "timed_out": False}

    delivered = int(outcome.get("delivered") or 0)
    if delivered:
        if outcome.get("timed_out"):
            return {
                "ok": False,
                "spoken": ["Your PC received the command but did not report back."],
                "status": "no_response",
                "action": "pc_relay",
                "pc_count": delivered,
            }
        spoken = outcome.get("spoken") or []
        return {
            # The PC's own words when it has any — that is where "This PC is
            # locked…" reaches the phone — and a plain confirmation otherwise.
            "ok": bool(outcome.get("ok")),
            "spoken": spoken or ["Done on your PC."],
            "status": "relayed",
            "action": "pc_relay",
            "pc_count": delivered,
        }

    # PC offline but the command was accepted onto the replay queue: say so
    # plainly. "Queued" and "recorded" are different promises — one will actually
    # run, the other never will — and the user has to be able to tell them apart.
    if outcome.get("queued"):
        depth = int(outcome.get("queue_depth") or 0)
        return {
            "ok": True,
            "spoken": ["Your PC is offline. I'll run this the moment it comes "
                       f"back online.{f' ({depth} waiting.)' if depth > 1 else ''}"],
            "status": "queued",
            "action": "pc_queue",
            "queue_depth": depth,
        }

    try:
        from modules.max_autonomy import execute_task
        # No PC connected AND not queueable (e.g. an unlock, which must never be
        # replayed later): run through the safety-gated coordinator in
        # record+sync mode — logged, risk-assessed and pushed to the AR HUD,
        # without faking hardware control.
        result = execute_task(
            cmd,
            source="guardian_command",
            device_name=device_name,
            context={"record_only": True},
        )
        spoken = list(result.response or [])
        return {"ok": bool(result.ok), "spoken": spoken, "status": result.status, "action": result.action}
    except Exception as exc:
        log.warning("guardian command routing failed: %s", exc)
        return {"ok": False, "spoken": [f"Command failed: {str(exc)[:160]}"]}


# --------------------------------------------------------------------------- #
# chat — the on-device JARVIS brain (Gemini-native response shape)
# --------------------------------------------------------------------------- #
def _gemini_envelope(text: str) -> dict[str, Any]:
    """Wrap plain text in the Gemini generateContent response shape the APK
    parses (candidates[].content.parts[].text + finishReason)."""
    return {
        "candidates": [
            {
                "content": {"parts": [{"text": text or ""}], "role": "model"},
                "finishReason": "STOP",
            }
        ]
    }


@router.post("/chat")
async def chat(request: Request) -> dict[str, Any]:
    """Synchronous LLM turn for the mobile app.

    Both the on-device autonomous automation loop (MobileAutomationEngine — reads
    the live screen via the Accessibility Service, asks JARVIS for the next
    CLICK/INPUT/SCROLL/DONE step) and the mobile JARVIS chat/decision engine
    (MobileJarvisEngine) POST {text, context} here and read a Gemini-native
    response (candidates[].content.parts[].text). The backend only ever served
    conversational chat at /agent/chat (async, {status:ok} over WebSocket), so
    /api/chat 404'd and the entire on-device brain was dead on the cloud.

    This runs the prompt synchronously through the token manager's full
    Gemini→NVIDIA fallback chain and returns the Gemini envelope, so both callers
    work with no APK change.
    """
    body = await _json_body(request)
    text = str(body.get("text") or "").strip()
    context = body.get("context") if isinstance(body.get("context"), dict) else {}
    persona = str(context.get("ai") or "jarvis").strip().lower() or "jarvis"
    if persona not in ("jarvis", "friday"):
        persona = "jarvis"
    if not text:
        return _gemini_envelope("Standing by.")
    try:
        from backend.services.token_manager import gemini_call_with_checkpoint
        reply = await gemini_call_with_checkpoint(text, task_type="general", persona=persona)
        return _gemini_envelope(reply or "")
    except Exception as exc:
        log.warning("mobile /api/chat LLM call failed: %s", exc)
        return _gemini_envelope("")


# --------------------------------------------------------------------------- #
# voice — TTS for the mobile assistant (WAV bytes)
# --------------------------------------------------------------------------- #
@router.get("/voice/speak")
async def voice_speak(text: str = "", voice: str = "friday", format: str = "wav") -> Response:
    """Mobile TTS. ContinuousVoiceRelay fetches `GET /api/voice/speak?text=&voice=
    &format=wav` and injects the raw PCM into its live WAV log. The backend only
    served TTS at `POST /voice/speak` (JSON body {text, agent}), so the mobile
    call 404'd on both path and method — JARVIS had no spoken voice on the cloud.
    Same XTTS-v2 pipeline as /voice/speak; `voice` maps to the persona."""
    text = (text or "").strip()
    if not text:
        return Response(status_code=204)
    try:
        from backend.voice.tts import TTSPipeline
        tts = TTSPipeline()
        audio = await tts.synthesize(text, personality=(voice or "friday"))
        return Response(content=audio, media_type="audio/wav")
    except Exception as exc:
        log.warning("mobile /api/voice/speak failed: %s", exc)
        return Response(status_code=500)


# --------------------------------------------------------------------------- #
# telemetry / events / autonomy
# --------------------------------------------------------------------------- #
@router.post("/phone_observe")
async def phone_observe(request: Request) -> dict[str, Any]:
    body = await _json_body(request)
    device_name = str(body.get("device_name") or "").strip()[:80]
    payload = {k: v for k, v in body.items() if k != "device_name"}
    try:
        from modules.max_autonomy import record_mobile_event
        record_mobile_event("phone_observe", payload, device_name=device_name)
    except Exception as exc:
        log.debug("phone_observe record failed: %s", exc)
        return {"ok": False, "error": str(exc)[:160]}
    return {"ok": True}


@router.post("/jarvis/mobile_event")
async def mobile_event(request: Request) -> dict[str, Any]:
    body = await _json_body(request)
    kind = str(body.get("kind") or "event").strip()[:80]
    device_name = str(body.get("device_name") or "").strip()[:80]
    payload = body.get("payload")
    if not isinstance(payload, dict):
        payload = {k: v for k, v in body.items() if k not in ("kind", "device_name", "ts")}
    try:
        from modules.max_autonomy import record_mobile_event
        record_mobile_event(kind, payload, device_name=device_name)
    except Exception as exc:
        log.debug("mobile_event record failed: %s", exc)
        return {"ok": False, "error": str(exc)[:160]}
    return {"ok": True}


@router.post("/max_autonomy/task")
async def max_autonomy_task(request: Request) -> dict[str, Any]:
    body = await _json_body(request)
    task = str(body.get("task") or "").strip()
    source = str(body.get("source") or "jarvis-native-android").strip()[:80]
    device_name = str(body.get("device_name") or "").strip()[:80]
    confirmed = bool(body.get("confirmed") or False)
    context = dict(body.get("context")) if isinstance(body.get("context"), dict) else {}
    # RUN IT ON THE PC IF THERE IS ONE, exactly as /command now does.
    #
    # The comment below was written when the Space genuinely had nowhere to send
    # a task, and it is no longer true: pc_relay_client.py holds an outbound
    # socket from the user's desktop and executes system:execute for real. With
    # record_only forced on the cloud, every Max Autonomy command from the phone
    # came back "Task recorded and synced" — which sounds like success, and is
    # why commands appeared to do nothing while the PC sat connected and idle.
    #
    # Falls through to record+sync when no desktop is listening, which is the
    # honest answer then rather than a claim of control.
    try:
        from backend.ws.agent_ws import ws_manager as _ws_manager
        if task and _ws_manager.pc_connections:
            outcome = await _ws_manager.execute_on_pcs(task, device_name=device_name)
            if outcome.get("delivered"):
                spoken = outcome.get("spoken") or []
                if outcome.get("timed_out"):
                    spoken = ["Your PC received the task but did not report back."]
                return {
                    "ok": bool(outcome.get("ok")),
                    "status": "relayed",
                    "task": task,
                    "action": "pc_relay",
                    "message": (spoken[0] if spoken else "Done on your PC."),
                    "requires_confirmation": False,
                    "blocked": False,
                    "risk": {},
                    "plan": [],
                    "response": spoken or ["Done on your PC."],
                }
    except Exception as exc:
        log.warning("autonomy PC relay unavailable: %s", exc)

    # No desktop listening. The headless Space has no PC of its own and does not
    # ship phone/command_runner, so autonomy runs in record+sync mode — logged,
    # risk-assessed and pushed to the AR HUD. Also the safer default: dangerous
    # "confirmed" actions record rather than fire.
    if os.environ.get("SPACE_ID"):
        context.setdefault("record_only", "true")
    try:
        from modules.max_autonomy import execute_task
        result = execute_task(
            task,
            source=source,
            device_name=device_name,
            context=context,
            confirmed=confirmed,
        )
        return result.as_dict()
    except Exception as exc:
        log.warning("max_autonomy task failed: %s", exc)
        return {
            "ok": False,
            "status": "error",
            "task": task,
            "action": "error",
            "message": f"Autonomy task failed: {str(exc)[:160]}",
            "requires_confirmation": False,
            "blocked": False,
            "risk": {},
            "plan": [],
            "response": [f"Autonomy task failed: {str(exc)[:160]}"],
        }