from __future__ import annotations import json import mimetypes import secrets import shutil import socket import subprocess import threading from dataclasses import dataclass from datetime import datetime, timedelta, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from hmac import compare_digest from ipaddress import ip_address from pathlib import Path from typing import Any from urllib.parse import parse_qs, urlencode, urlparse from adam.generations import ( ChatGenerationRequest, build_generation_plan, generation_model_match_score, generation_tools, load_generation_history, parse_chat_generation_request, ) from adam.remote_dispatcher import RemoteCommandDispatcher from adam.remote_media import OpaqueIdCodec, RemoteMediaStore from adam.remote_v1 import RemoteV1Service REMOTE_MODE_DISABLED = "disabled" REMOTE_MODE_LOCAL = "local_wifi" REMOTE_MODE_TAILSCALE = "tailscale" class RemoteHTTPServer(ThreadingHTTPServer): """Bound concurrent connections so slow clients cannot spawn unlimited threads.""" allow_reuse_address = True daemon_threads = True def __init__(self, *args, **kwargs): self._slots = threading.BoundedSemaphore(16) super().__init__(*args, **kwargs) def process_request(self, request, client_address): if not self._slots.acquire(blocking=False): self.shutdown_request(request) return try: super().process_request(request, client_address) except BaseException: self._slots.release() raise def process_request_thread(self, request, client_address): try: super().process_request_thread(request, client_address) finally: self._slots.release() @dataclass(frozen=True, slots=True) class TailscaleStatus: installed: bool = False connected: bool = False device_name: str = "" dns_name: str = "" tailscale_ip: str = "" backend_state: str = "" serve_available: bool = False serve_running: bool = False message: str = "Tailscale is not installed." def default_remote_settings() -> dict[str, Any]: return { "enabled": False, "remote_mode": REMOTE_MODE_LOCAL, "bind_address": "127.0.0.1", "port": 8765, "token": secrets.token_urlsafe(24), "allow_job_control": False, "auto_approve_training": False, } def remote_scope(bind_address: str) -> str: bind = bind_address.strip().casefold() if bind == "localhost": return "local-device only" if bind in {"0.0.0.0", "::"}: return "all network interfaces" try: address = ip_address(bind.strip("[]")) except ValueError: return "custom bind address" if address.is_loopback: return "local-device only" if address.is_private or address.is_link_local: return "local network" return "custom bind address" def local_network_host() -> str: """Best-effort address other devices on the same network can use.""" try: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: sock.connect(("8.8.8.8", 80)) host = str(sock.getsockname()[0]) except OSError: try: host = socket.gethostbyname(socket.gethostname()) except OSError: return "" try: address = ip_address(host) except ValueError: return "" if address.is_loopback or address.is_unspecified: return "" return host if address.is_private or address.is_link_local else "" def inspect_tailscale( runner: Any | None = None, which: Any | None = None, ) -> TailscaleStatus: which = which or shutil.which executable = which("tailscale") if not executable: return TailscaleStatus() runner = runner or _run_tailscale try: status = runner([executable, "status", "--json"]) except OSError as exc: return TailscaleStatus(installed=True, message=f"Tailscale could not be checked: {exc}") if getattr(status, "returncode", 1) != 0: error = _command_text(getattr(status, "stderr", "")) or "Tailscale is installed but not connected." return TailscaleStatus(installed=True, message=error) try: payload = json.loads(_command_text(getattr(status, "stdout", "")) or "{}") except json.JSONDecodeError: return TailscaleStatus(installed=True, message="Tailscale returned an unreadable status response.") self_node = payload.get("Self") if isinstance(payload, dict) else {} self_node = self_node if isinstance(self_node, dict) else {} ips = [str(item) for item in self_node.get("TailscaleIPs", []) if str(item)] backend = str(payload.get("BackendState", "") or "") connected = backend.casefold() == "running" or bool(ips) serve_status = _tailscale_serve_running(runner, executable) return TailscaleStatus( installed=True, connected=connected, device_name=str(self_node.get("HostName", "") or ""), dns_name=str(self_node.get("DNSName", "") or "").rstrip("."), tailscale_ip=next((ip for ip in ips if "." in ip), ips[0] if ips else ""), backend_state=backend, serve_available=serve_status is not None, serve_running=bool(serve_status), message="Tailscale is connected." if connected else "Tailscale is installed but disconnected.", ) def _tailscale_serve_running(runner: Any, executable: str) -> bool | None: try: result = runner([executable, "serve", "status", "--json"]) except OSError: return None if getattr(result, "returncode", 1) != 0: return None text = _command_text(getattr(result, "stdout", "")).strip() return bool(text and text not in {"{}", "null"}) def _run_tailscale(command: list[str]) -> subprocess.CompletedProcess[str]: return subprocess.run(command, capture_output=True, text=True, timeout=8, check=False) def _command_text(value: Any) -> str: if isinstance(value, bytes): return value.decode("utf-8", errors="replace") return str(value or "") def _remote_prompt_from_payload(payload: dict[str, Any]) -> str: for key in ("prompt", "message", "text", "request", "input"): value = payload.get(key) if isinstance(value, str) and value.strip(): return value.strip() return "" def _remote_dashboard_html() -> str: return _remote_dashboard_app_html() return """ ADAM Remote
ADAM AI Development and Automation Manager
Connecting

Prompt ADAM

Plans that need desktop approval will wait safely inside the main ADAM app.

Generate

Generation uses ADAM's existing desktop backend and queue.

Active Job

Checking ADAM...
Waiting for status.
...
--
--
--
--

Live Preview

Latest ADAM preview
The latest training or generation preview will appear here.

Waiting for preview output.

Latest Generation

Latest generated ADAM image
Finished generated images will appear here.

Waiting for a completed generation.

System

--
--
--
--

Remote Control

--

Status-only remote access is loading.

Remote settings are synced from ADAM.

Recent Queue

Completed Jobs

Failed Jobs

""" def _remote_dashboard_app_html() -> str: from adam.remote_dashboard import remote_dashboard_app_html return remote_dashboard_app_html() return """ ADAM Remote
ADAM Remote
Connecting...
Active Job
No active job.
Live Preview
Live Preview
Waiting for a preview.
Latest Generation
Prompt ADAM
Structured Training

    

  

  

    
Generate Image

      
      
      
      
      
      

      

      

      

    

  

  

    
Datasets

    
    
    
Models

  

  

    
System

    
Remote Control

    
Jobs

  






"""


class RemoteAccessService:
    """Small authenticated local API foundation for browser/device clients."""

    def __init__(self, config: Any, jobs: Any, monitor: Any, planner: Any = None) -> None:
        self.config = config
        self.jobs = jobs
        self.monitor = monitor
        self.planner = planner
        self.root = Path(getattr(config, "root", None) or getattr(planner, "root", None) or Path.cwd()).resolve()
        self.dispatcher = RemoteCommandDispatcher()
        token = ""
        try:
            token = str(self.settings().get("token", ""))
        except Exception:
            token = ""
        self.codec = OpaqueIdCodec(f"{self.root}|{token}")
        self.media = RemoteMediaStore(self.root, self.codec)
        self.api_v1 = RemoteV1Service(
            root=self.root,
            config=config,
            jobs=jobs,
            planner=planner,
            dispatcher=self.dispatcher,
            codec=self.codec,
            media=self.media,
            auto_approve_training=self._should_auto_approve_training,
        )
        self._server: ThreadingHTTPServer | None = None
        self._thread: threading.Thread | None = None

    @property
    def running(self) -> bool:
        return self._server is not None

    def settings(self) -> dict[str, Any]:
        values = default_remote_settings()
        stored = self.config.get("remote_access", {})
        if isinstance(stored, dict):
            values.update(stored)
        if not isinstance(stored, dict) or not stored.get("token"):
            values["token"] = secrets.token_urlsafe(24)
            self.config.update({"remote_access": values})
        return values

    def save_settings(self, values: dict[str, Any]) -> None:
        clean = self.settings()
        port = clean["port"]
        try:
            port = int(values.get("port", clean["port"]))
        except (TypeError, ValueError):
            port = clean["port"]
        clean.update(
            {
                "enabled": bool(values.get("enabled", clean["enabled"])),
                "remote_mode": self._clean_mode(str(values.get("remote_mode", clean["remote_mode"]))),
                "bind_address": str(values.get("bind_address", clean["bind_address"])).strip() or "127.0.0.1",
                "port": max(1024, min(port, 65535)),
                "token": str(values.get("token", clean["token"])).strip() or secrets.token_urlsafe(24),
                "allow_job_control": bool(values.get("allow_job_control", clean["allow_job_control"])),
                "auto_approve_training": bool(values.get("auto_approve_training", clean["auto_approve_training"])),
            }
        )
        self.config.update({"remote_access": clean})

    def start(self) -> str:
        if self.running:
            return self.url()
        settings = self.settings()
        mode = self._clean_mode(str(settings.get("remote_mode", REMOTE_MODE_LOCAL)))
        if mode == REMOTE_MODE_DISABLED:
            raise RuntimeError("Remote access is disabled by Remote Mode.")
        if not settings.get("enabled"):
            raise RuntimeError("Remote access is disabled.")
        bind = "127.0.0.1" if mode == REMOTE_MODE_TAILSCALE else str(settings["bind_address"])
        port = int(settings["port"])
        jobs = self.jobs
        monitor = self.monitor
        service = self
        api_v1 = self.api_v1

        class Handler(BaseHTTPRequestHandler):
            def setup(self) -> None:
                self.request.settimeout(10)
                super().setup()

            def _authorized(self) -> bool:
                current = service.settings()
                if not current.get("enabled") or current.get("remote_mode") == REMOTE_MODE_DISABLED:
                    return False
                token = str(current["token"]).encode("utf-8")
                header = self.headers.get("Authorization", "")
                query_token = ""
                parsed = urlparse(self.path)
                values = parse_qs(parsed.query)
                if values.get("token"):
                    query_token = values["token"][0]
                bearer = header.removeprefix("Bearer ").strip()
                return compare_digest(bearer.encode("utf-8"), token) or (
                    bool(query_token) and self._browser_token_allowed() and compare_digest(query_token.encode("utf-8"), token)
                )

            def _security_headers(self) -> None:
                self.send_header("X-Content-Type-Options", "nosniff")
                self.send_header("Referrer-Policy", "no-referrer")
                self.send_header("X-Frame-Options", "DENY")
                self.send_header("Content-Security-Policy", "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; img-src 'self' data: blob:; connect-src 'self'; object-src 'none'; base-uri 'none'; frame-ancestors 'none'; form-action 'self'")

            def _valid_post(self) -> bool:
                origin = self.headers.get("Origin")
                if self.headers.get("Sec-Fetch-Site") == "cross-site" or (origin and (
                    urlparse(origin).scheme not in {"http", "https"}
                    or urlparse(origin).netloc.casefold() != self.headers.get("Host", "").casefold()
                )):
                    self._send(403, {"error": "Cross-site requests are not allowed."})
                    return False
                if self.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() != "application/json":
                    self._send(415, {"error": "Use application/json for remote commands."})
                    return False
                lengths = self.headers.get_all("Content-Length", [])
                try:
                    length = int(lengths[0]) if len(lengths) == 1 else -1
                except ValueError:
                    length = -1
                if self.headers.get("Transfer-Encoding") or length < 0 or length > 20_000:
                    self._send(413 if length > 20_000 else 400, {"error": "Invalid request size (maximum 20000 bytes)."})
                    return False
                return True

            def _browser_token_allowed(self) -> bool:
                try:
                    client = ip_address(str(self.client_address[0]).strip("[]"))
                except ValueError:
                    return False
                if client.is_loopback:
                    return True
                return remote_scope(bind) != "local-device only" and (client.is_private or client.is_link_local)

            def _send(self, status: int, payload: dict[str, Any]) -> None:
                body = json.dumps(payload).encode("utf-8")
                self._send_bytes(status, body, "application/json")

            def _send_html(self, status: int, html: str) -> None:
                self._send_bytes(status, html.encode("utf-8"), "text/html; charset=utf-8")

            def _send_bytes(self, status: int, body: bytes, content_type: str) -> None:
                if status >= 400 and self.command == "POST" and not getattr(self, "_body_read", False):
                    # Drain a small, already-sent body before closing; Windows can
                    # otherwise reset the connection before the error is delivered.
                    try:
                        length = int(self.headers.get("Content-Length", "0"))
                        if 0 < length <= 20_000 and not self.headers.get("Transfer-Encoding"):
                            self.connection.settimeout(0.25)
                            self.rfile.read(length)
                    except (ValueError, OSError):
                        pass
                    finally:
                        self.connection.settimeout(10)
                        self._body_read = True
                try:
                    self.send_response(status)
                    self.send_header("Content-Type", content_type)
                    self.send_header("Cache-Control", "no-store")
                    self._security_headers()
                    self.send_header("Content-Length", str(len(body)))
                    self.end_headers()
                    self.wfile.write(body)
                except (BrokenPipeError, ConnectionAbortedError, ConnectionResetError):
                    return

            def do_GET(self) -> None:
                if not self._authorized():
                    self._send(401, {"error": "Missing or invalid remote access token."})
                    return
                path = urlparse(self.path).path
                response = api_v1.route("GET", path, urlparse(self.path).query)
                if response is not None:
                    self._send_response(response)
                    return
                if path == "/":
                    self._send_html(200, _remote_dashboard_html())
                    return
                if path == "/api/preview":
                    self._send_preview()
                    return
                if path == "/api/generation-image":
                    self._send_generation_image()
                    return
                if path != "/api/status":
                    self._send(404, {"error": "Unknown endpoint."})
                    return
                self._send(200, service._status_payload(bind, service.settings()))

            def do_POST(self) -> None:
                if not self._authorized():
                    self._send(401, {"error": "Missing or invalid remote access token."})
                    return
                if not self._valid_post():
                    return
                path = urlparse(self.path).path
                payload = None
                if path.startswith("/api/v1/"):
                    payload = self._read_json_body()
                    if payload is None:
                        self._send(400, {"error": "Send a valid JSON object."})
                        return
                    response = api_v1.route("POST", path, urlparse(self.path).query, payload)
                    if response is not None:
                        self._send_response(response)
                        return
                if path == "/api/job":
                    self._handle_job_action()
                    return
                if path == "/api/remote-settings":
                    self._handle_remote_settings()
                    return
                if path != "/api/prompt":
                    self._send(404, {"error": "Unknown endpoint."})
                    return
                payload = self._read_json_body()
                if payload is None:
                    self._send(400, {"error": "Send a valid prompt."})
                    return
                prompt = _remote_prompt_from_payload(payload)
                if not prompt:
                    self._send(400, {"error": "Type a prompt for ADAM first."})
                    return
                if len(prompt) > 2_000:
                    self._send(400, {"error": "Keep remote prompts under 2000 characters."})
                    return
                result = service.submit_prompt(prompt)
                self._send(200 if result.get("ok") else 400, result)

            def _send_response(self, response: Any) -> None:
                try:
                    self.send_response(int(response.status))
                    self.send_header("Content-Type", str(response.content_type))
                    headers = response.headers or {}
                    if "Cache-Control" in headers:
                        self.send_header("Cache-Control", headers["Cache-Control"])
                    else:
                        self.send_header("Cache-Control", "no-store")
                    self._security_headers()
                    self.send_header("Content-Length", str(len(response.body)))
                    self.end_headers()
                    self.wfile.write(response.body)
                except (BrokenPipeError, ConnectionAbortedError, ConnectionResetError):
                    return

            def _read_json_body(self) -> dict[str, Any] | None:
                self._body_read = True
                try:
                    length = int(self.headers.get("Content-Length", "0") or "0")
                except ValueError:
                    length = 0
                try:
                    payload = json.loads(self.rfile.read(length).decode("utf-8")) if length else {}
                except (UnicodeDecodeError, json.JSONDecodeError, OSError):
                    return None
                return payload if isinstance(payload, dict) else None

            def _handle_job_action(self) -> None:
                payload = self._read_json_body()
                if payload is None:
                    self._send(400, {"error": "Send a valid job action."})
                    return
                action = str(payload.get("action", "")).strip().casefold()
                job_id = str(payload.get("job_id", "")).strip()
                result = service.job_action(job_id, action, bool(service.settings().get("allow_job_control")))
                self._send(200 if result.get("ok") else 400, result)

            def _handle_remote_settings(self) -> None:
                payload = self._read_json_body()
                if payload is None:
                    self._send(400, {"error": "Send valid remote settings."})
                    return
                current = service.settings()
                enabled = payload.get("auto_approve_training", False)
                if not isinstance(enabled, bool):
                    self._send(400, {"error": "Auto-approval must be true or false."})
                    return
                if enabled and not current.get("allow_job_control"):
                    self._send(403, {"error": "Enable remote job controls in the desktop app before changing approval permissions."})
                    return
                def update_approval():
                    # Recheck on the owning thread and only change this permission.
                    # A queued request must not restore an older token or settings.
                    if enabled and not service.settings().get("allow_job_control"):
                        return False
                    service.save_settings({"auto_approve_training": enabled})
                    return True

                if not service.dispatcher.call_ui(update_approval):
                    self._send(403, {"error": "Remote job controls have been disabled."})
                    return
                state = "on" if enabled else "off"
                self._send(200, {"ok": True, "message": f"Auto-approval is {state}."})

            def _send_preview(self) -> None:
                path = service.preview_path()
                if path is None or not path.is_file():
                    self._send(404, {"error": "No preview image is available yet."})
                    return
                content_type = mimetypes.guess_type(str(path))[0] or "image/png"
                try:
                    body = path.read_bytes()
                except OSError:
                    self._send(404, {"error": "Preview image is no longer available."})
                    return
                self._send_bytes(200, body, content_type)

            def _send_generation_image(self) -> None:
                parsed = urlparse(self.path)
                values = parse_qs(parsed.query)
                try:
                    record_index = int(values.get("record", ["0"])[0])
                    image_index = int(values.get("image", ["0"])[0])
                except (TypeError, ValueError):
                    self._send(400, {"error": "Choose a valid generation image."})
                    return
                path = service.generation_image_path(record_index, image_index)
                if path is None or not path.is_file():
                    self._send(404, {"error": "Generated image is no longer available."})
                    return
                content_type = mimetypes.guess_type(str(path))[0] or "image/png"
                try:
                    body = path.read_bytes()
                except OSError:
                    self._send(404, {"error": "Generated image is no longer available."})
                    return
                self._send_bytes(200, body, content_type)

            def log_message(self, _format: str, *_args: Any) -> None:
                return

        self._server = RemoteHTTPServer((bind, port), Handler)
        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
        self._thread.start()
        return self.url()

    def _status_payload(self, bind: str, settings: dict[str, Any]) -> dict[str, Any]:
        snapshot = self.monitor.snapshot() if self.monitor is not None else None
        active = self.jobs.active_job if self.jobs is not None else None
        return {
            "app": "ADAM",
            "scope": remote_scope(bind),
            "permissions": {
                "status": True,
                "system": True,
                "queue_view": True,
                "prompt": self.planner is not None and self.jobs is not None,
                "job_control": bool(settings.get("allow_job_control")),
                "auto_approve_training": bool(settings.get("auto_approve_training")),
                "dangerous_actions": False,
            },
            "active_job": None
            if active is None
            else {
                "id": active.id,
                "project": active.plan.project_name,
                "status": active.status.value,
                "progress": active.progress,
                "timing": self._job_timing(active),
                "preview": self._job_preview(active),
            },
            "preview": self.preview_payload(),
            "latest_generation": self.latest_generation_payload(),
            "queue": [
                self._job_summary(job)
                for job in (self.jobs.jobs[:20] if self.jobs is not None else [])
                if job.status.value not in {"Finished", "Failed", "Cancelled"}
            ],
            "completed_jobs": [
                self._job_summary(job)
                for job in (self.jobs.jobs[:30] if self.jobs is not None else [])
                if job.status.value == "Finished"
            ],
            "failed_jobs": [
                self._job_summary(job)
                for job in (self.jobs.jobs[:30] if self.jobs is not None else [])
                if job.status.value in {"Failed", "Cancelled", "Interrupted"}
            ],
            "system": {
                "cpu_percent": snapshot.cpu_percent if snapshot else None,
                "memory_percent": snapshot.memory_percent if snapshot else None,
                "gpu_name": snapshot.gpu_name if snapshot else "",
                "gpu_percent": snapshot.gpu_percent if snapshot else None,
                "vram_percent": snapshot.vram_percent if snapshot else None,
                "gpu_temperature": snapshot.gpu_temperature if snapshot else None,
            },
        }

    @staticmethod
    def _job_summary(job: Any) -> dict[str, Any]:
        try:
            current_step = int(getattr(job, "current_step", -1))
        except (TypeError, ValueError):
            current_step = -1
        steps = list(getattr(getattr(job, "plan", None), "steps", []) or [])
        current_step_title = ""
        if 0 <= current_step < len(steps):
            current_step_title = str(getattr(steps[current_step], "title", "") or "")
        return {
            "id": getattr(job, "id", ""),
            "project": getattr(getattr(job, "plan", None), "project_name", ""),
            "status": getattr(getattr(job, "status", None), "value", str(getattr(job, "status", ""))),
            "progress": getattr(job, "progress", 0),
            "timing": RemoteAccessService._job_timing(job),
            "error": getattr(job, "error", "") or "",
            "started_at": getattr(job, "started_at", "") or "",
            "ended_at": getattr(job, "ended_at", "") or "",
            "scheduled_for": getattr(job, "scheduled_for", "") or "",
            "current_step": current_step,
            "step_count": len(steps),
            "current_step_title": current_step_title,
            "requires_confirmation": bool(getattr(getattr(job, "plan", None), "requires_confirmation", False)),
            "output_folder": Path(str(getattr(job, "output_folder", "") or "")).name,
            "progress_current": getattr(job, "progress_current", 0) or 0,
            "progress_total": getattr(job, "progress_total", 0) or 0,
            "progress_unit": getattr(job, "progress_unit", "") or "",
            "logs": list(getattr(job, "logs", []) or [])[-3:],
        }

    @staticmethod
    def _job_timing(job: Any) -> dict[str, Any]:
        started = RemoteAccessService._parse_time(getattr(job, "started_at", "") or "")
        ended = RemoteAccessService._parse_time(getattr(job, "ended_at", "") or "")
        progress = max(0, min(100, int(getattr(job, "progress", 0) or 0)))
        status = getattr(getattr(job, "status", None), "value", str(getattr(job, "status", "")))
        review = getattr(getattr(job, "plan", None), "orion_review", {}) or {}
        try:
            estimate_seconds = int(float(review.get("estimated_high_minutes", 0) or 0) * 60)
        except (TypeError, ValueError):
            estimate_seconds = 0

        now = datetime.now(timezone.utc)
        elapsed_seconds = 0
        if started is not None:
            finish = ended or now
            elapsed_seconds = max(0, int((finish - started).total_seconds()))

        remaining_seconds: int | None = None
        basis = ""
        if status in {"Finished", "Failed", "Cancelled", "Interrupted"}:
            remaining_seconds = 0
            basis = "complete"
        elif started is not None and progress > 0:
            remaining_seconds = max(0, int(elapsed_seconds * (100 - progress) / progress))
            basis = "progress"
        elif estimate_seconds:
            remaining_seconds = max(0, estimate_seconds - elapsed_seconds)
            basis = "planning estimate"

        finish_label = ""
        if remaining_seconds is not None and remaining_seconds > 0:
            finish_label = (now + timedelta(seconds=remaining_seconds)).astimezone().strftime("%I:%M %p").lstrip("0")

        return {
            "elapsed_seconds": elapsed_seconds if started is not None else None,
            "remaining_seconds": remaining_seconds,
            "estimated_total_seconds": estimate_seconds or None,
            "elapsed_label": RemoteAccessService._format_duration(elapsed_seconds) if started is not None else "",
            "remaining_label": (
                "done" if remaining_seconds == 0 and basis == "complete"
                else f"about {RemoteAccessService._format_duration(remaining_seconds)}" if remaining_seconds is not None else ""
            ),
            "finish_label": finish_label,
            "estimate_label": f"up to {RemoteAccessService._format_duration(estimate_seconds)}" if estimate_seconds else "",
            "basis": basis,
        }

    @staticmethod
    def _parse_time(value: str) -> datetime | None:
        if not value:
            return None
        try:
            parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
        except ValueError:
            return None
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=timezone.utc)
        return parsed.astimezone(timezone.utc)

    @staticmethod
    def _format_duration(seconds: int | float | None) -> str:
        if seconds is None:
            return ""
        total = max(0, int(seconds))
        if total < 60:
            return f"{total}s"
        minutes, sec = divmod(total, 60)
        if minutes < 60:
            return f"{minutes}m {sec}s" if sec else f"{minutes}m"
        hours, minute = divmod(minutes, 60)
        if hours < 24:
            return f"{hours}h {minute}m" if minute else f"{hours}h"
        days, hour = divmod(hours, 24)
        return f"{days}d {hour}h" if hour else f"{days}d"

    def preview_payload(self) -> dict[str, Any]:
        if self.jobs is None:
            return {"available": False}
        candidates = []
        if self.jobs.active_job is not None:
            candidates.append(self.jobs.active_job)
        candidates.extend(self.jobs.jobs[:20])
        for job in candidates:
            preview = self._job_preview(job)
            if preview["available"]:
                return preview
        return {"available": False}

    def preview_path(self) -> Path | None:
        if self.jobs is None:
            return None
        candidates = []
        if self.jobs.active_job is not None:
            candidates.append(self.jobs.active_job)
        candidates.extend(self.jobs.jobs[:20])
        for job in candidates:
            path = self._job_preview_path(job)
            if path is not None:
                return path
        return None

    @staticmethod
    def _job_preview(job: Any) -> dict[str, Any]:
        if RemoteAccessService._job_preview_path(job) is None:
            return {"available": False}
        return {
            "available": True,
            "url": "/api/preview",
            "kind": getattr(job, "preview_kind", ""),
            "epoch": getattr(job, "preview_epoch", 0),
            "current": getattr(job, "preview_current", 0),
            "total": getattr(job, "preview_total", 0),
            "prompt": getattr(job, "preview_prompt", ""),
        }

    @staticmethod
    def _job_preview_path(job: Any) -> Path | None:
        path = str(getattr(job, "preview_path", "") or "")
        if not path:
            return None
        target = Path(path)
        return target if target.is_file() else None

    def latest_generation_payload(self) -> dict[str, Any]:
        records = self._generation_records()
        if not records:
            return {"available": False}
        record = records[0]
        return {
            "available": True,
            "url": "/api/generation-image?record=0&image=0",
            "images": [
                {
                    "index": index,
                    "url": f"/api/generation-image?record=0&image={index}",
                }
                for index, _path in enumerate(record.images)
            ],
            "model_name": record.model_name,
            "provider_id": record.provider_id,
            "provider_name": record.provider_name,
            "prompt": record.prompt,
            "seed": record.seed,
            "steps": record.steps,
            "sampler": record.sampler,
            "aspect_ratio": record.aspect_ratio,
            "created_at": record.created_at,
            "image_count": len(record.images),
        }

    def generation_image_path(self, record_index: int, image_index: int) -> Path | None:
        if record_index < 0 or image_index < 0:
            return None
        records = self._generation_records(limit=max(1, record_index + 1))
        if record_index >= len(records):
            return None
        record = records[record_index]
        if image_index >= len(record.images):
            return None
        path = record.images[image_index]
        return path if path.is_file() else None

    def _generation_records(self, *, limit: int = 30):
        root = self._generation_root()
        if root is None:
            return []
        return load_generation_history(root, limit=limit)

    def _generation_root(self) -> Path | None:
        for candidate in (
            getattr(self.planner, "root", None),
            getattr(getattr(self.planner, "registry", None), "root", None),
        ):
            if candidate:
                return Path(candidate)
        return None

    def submit_prompt(self, prompt: str) -> dict[str, Any]:
        generation_result = self._submit_generation_prompt(prompt)
        if generation_result is not None:
            return generation_result
        if self.planner is None or self.jobs is None:
            return {"ok": False, "error": "Remote prompting is not available in this ADAM session."}
        try:
            def prepare_plan():
                from adam.training_assistant import append_preflight_summary

                plan = self.planner.plan(prompt)
                append_preflight_summary(plan, self.config)
                return plan

            plan = self.dispatcher.call_background(prepare_plan)
        except Exception as exc:
            return {"ok": False, "error": f"ADAM could not plan that request: {exc}"}
        if not plan.steps:
            return {"ok": True, "message": plan.summary or "ADAM received your message.", "requires_approval": False}
        job = self.dispatcher.submit_job(self.jobs, plan)
        if self._should_auto_approve_training(plan):
            self.dispatcher.confirm_job(self.jobs, job.id)
            return {
                "ok": True,
                "message": f"Queued {job.plan.project_name}. Remote training auto-approval is on.",
                "job_id": job.id,
                "requires_approval": False,
                "auto_approved": True,
            }
        if plan.requires_confirmation:
            return {
                "ok": True,
                "message": f"Plan created for {job.plan.project_name}. It needs approval in the desktop app before it runs.",
                "job_id": job.id,
                "requires_approval": True,
            }
        return {
            "ok": True,
            "message": f"Queued {job.plan.project_name}.",
            "job_id": job.id,
            "requires_approval": False,
        }

    def job_action(self, job_id: str, action: str, allowed: bool) -> dict[str, Any]:
        if not allowed:
            return {"ok": False, "error": "Remote job controls are disabled in ADAM."}
        if self.jobs is None:
            return {"ok": False, "error": "Job controls are not available in this ADAM session."}
        if not job_id:
            return {"ok": False, "error": "Choose a job first."}
        if action not in {"cancel", "retry", "confirm", "pause", "resume", "end"}:
            return {"ok": False, "error": "Unsupported remote job action."}
        try:
            result = self.dispatcher.job_action(self.jobs, job_id, action)
            job = result.get("job")
            if action == "cancel":
                return {"ok": True, "message": f"Cancellation requested for {job.plan.project_name}."}
            if action == "confirm":
                return {"ok": True, "message": f"Approved {job.plan.project_name}."}
            if action == "pause":
                return {"ok": True, "message": f"Paused {job.plan.project_name}."}
            if action == "resume":
                return {"ok": True, "message": f"Resumed {job.plan.project_name}."}
            if action == "end":
                return {"ok": True, "message": f"Ended {job.plan.project_name}."}
            retried = result.get("retried")
            return {"ok": True, "message": f"Retry queued for {retried.plan.project_name}.", "job_id": retried.id}
        except Exception as exc:
            return {"ok": False, "error": f"ADAM could not update that job: {exc}"}

    def _submit_generation_prompt(self, prompt: str) -> dict[str, Any] | None:
        parsed = parse_chat_generation_request(prompt)
        if parsed is None:
            return None
        if self.planner is None or self.jobs is None:
            return {"ok": False, "error": "Remote image generation is not available in this ADAM session."}
        if not all(hasattr(self.planner, name) for name in ("assets", "registry")):
            return {"ok": False, "error": "Remote image generation needs the full ADAM planner session."}
        try:
            plan = self._generation_plan(parsed)
        except ValueError as exc:
            return {"ok": False, "error": str(exc)}
        try:
            job = self.dispatcher.submit_job(self.jobs, plan)
        except Exception as exc:
            return {"ok": False, "error": f"ADAM could not queue that generation: {exc}"}
        return {
            "ok": True,
            "message": f"Queued {job.plan.project_name}.",
            "job_id": job.id,
            "requires_approval": bool(job.plan.requires_confirmation),
        }

    def _should_auto_approve_training(self, plan: Any) -> bool:
        if getattr(plan, "orion_review", {}).get("level") == "warning":
            return False
        if not getattr(plan, "requires_confirmation", False):
            return False
        if not bool(self.settings().get("auto_approve_training")):
            return False
        return any(
            str(getattr(step, "tool_id", "")).endswith("_trainer")
            for step in getattr(plan, "steps", [])
        )

    def _generation_plan(self, parsed: ChatGenerationRequest):
        assets = self.planner.assets
        registry = self.planner.registry
        if hasattr(assets, "discover"):
            assets.discover(self.config)
        tools = generation_tools(registry)
        if not tools:
            raise ValueError("No image generators are currently available in ADAM.")

        stable_diffusion_request = (
            parsed.has_positive_prompt
            or bool(parsed.base_model_query)
            or bool(parsed.negative_prompt)
            or parsed.cfg_scale is not None
            or parsed.lora_strength is not None
            or parsed.denoise_strength is not None
        )
        plain_model_search = (
            not parsed.provider_hint
            and not stable_diffusion_request
            and not parsed.model_query
        )
        base_only = (
            stable_diffusion_request
            and parsed.provider_hint != "lora"
            and not parsed.model_query
        )
        preferred_id = {
            "ddpm": "ddpm_generator",
            "flow": "flow_generator",
            "lora": "lora_generator",
        }.get(parsed.provider_hint, "")
        if stable_diffusion_request and parsed.provider_hint not in {"ddpm", "flow"}:
            preferred_id = "lora_generator"
        preferred_tool = next((item for item in tools if item.id == preferred_id), None)
        if parsed.provider_hint and preferred_tool is None:
            raise ValueError(f"The requested {parsed.provider_hint.upper()} image generator is not currently available.")

        candidate_tools = [preferred_tool] if preferred_tool else tools
        candidates = [
            asset
            for asset in getattr(assets, "assets", [])
            if asset.kind == "model"
            and (not plain_model_search or asset.trainer in {"ddpm", "flow"})
            and not (plain_model_search and parsed.reference_image and asset.trainer == "flow")
            and any(
                item is not None and asset.trainer in item.model_trainers
                for item in candidate_tools
            )
            and self._generation_model_is_ready(asset)
        ]
        model_query = parsed.model_query or (parsed.subject if not base_only else "")
        scored = sorted(
            (
                (generation_model_match_score(model_query, asset.name), asset)
                for asset in candidates
            ),
            key=lambda item: item[0],
            reverse=True,
        )
        model = scored[0][1] if scored and scored[0][0] > 0 else None
        if model is None and not model_query and len(candidates) == 1:
            model = candidates[0]
        if model is None and plain_model_search:
            base_only = True
            preferred_tool = next((item for item in tools if item.id == "lora_generator"), None)
            candidate_tools = [preferred_tool] if preferred_tool else tools
            model_query = ""
        if model is None and not base_only:
            detail = f' matching "{model_query}"' if model_query else ""
            examples: list[str] = []
            for asset in candidates:
                if asset.name not in examples:
                    examples.append(asset.name)
                if len(examples) == 4:
                    break
            available = f" Available examples: {', '.join(examples)}." if examples else ""
            raise ValueError(
                f"I could not find a completed image model{detail}.{available} "
                'Try: Generate an image using model "Model Name".'
            )

        tool = next(
            (
                item
                for item in candidate_tools
                if item is not None and (base_only or model.trainer in item.model_trainers)
            ),
            None,
        )
        if tool is None:
            raise ValueError("The matching model does not have an available image generator.")
        if parsed.reference_image and "reference_image" not in tool.capabilities:
            raise ValueError(
                f"{tool.name} does not support reference-image conditioning. "
                "Remove the attachment or choose LoRA/Stable Diffusion or DDPM."
            )

        options = tool.generation_options
        saved_generation = self.config.get("generation_settings", {})
        saved_generation = saved_generation if isinstance(saved_generation, dict) else {}
        sampler_options = [str(value) for value in options.get("samplers", [])]
        sampler = parsed.sampler or (
            str(saved_generation.get("sampler", "")) if tool.id == "lora_generator" else ""
        )
        if sampler not in sampler_options:
            sampler = sampler_options[0] if sampler_options else sampler or "DDIM"
        aspect_options = [str(value) for value in options.get("aspect_ratios", [])]
        aspect = parsed.aspect_ratio or (
            str(saved_generation.get("aspect", "")) if tool.id == "lora_generator" else ""
        )
        if aspect and aspect not in aspect_options:
            aspect = next(
                (value for value in aspect_options if value.startswith(f"{aspect} ") or value == aspect),
                "",
            )
        if not aspect:
            aspect = aspect_options[0] if aspect_options else "1:1 (Square)"
        step_min = int(options.get("step_min", 1) or 1)
        step_max = int(options.get("step_max", 500) or 500)
        default_steps = (
            saved_generation.get("steps", options.get("step_default", 50))
            if tool.id == "lora_generator"
            else options.get("step_default", 50)
        )
        steps = parsed.steps if parsed.steps is not None else int(default_steps or 50)
        steps = max(step_min, min(steps, step_max))
        count_limit = 8 if tool.id == "lora_generator" else 32
        default_count = (
            int(saved_generation.get("images", 1) or 1)
            if tool.id == "lora_generator"
            else 1
        )
        count = max(1, min(parsed.image_count or default_count, count_limit))
        seed = parsed.seed if parsed.seed is not None else 0
        extra_arguments: dict[str, Any] = {}
        if tool.id == "ddpm_generator":
            extra_arguments = {
                "reference_image": parsed.reference_image,
                "reference_strength": max(
                    0,
                    min(parsed.reference_strength if parsed.reference_strength is not None else 65, 100),
                ),
                "width": 0,
                "height": 0,
            }
        elif tool.id == "lora_generator":
            base_model_path = self._stable_diffusion_base_model_path(parsed, saved_generation)
            extra_arguments = {
                "negative_prompt": parsed.negative_prompt or str(saved_generation.get("negative_prompt", "")),
                "base_model_path": base_model_path,
                "width": 0,
                "height": 0,
                "cfg_scale": parsed.cfg_scale if parsed.cfg_scale is not None else float(saved_generation.get("cfg_scale", 0) or 0),
                "lora_strength": 0.0 if base_only else (
                    parsed.lora_strength if parsed.lora_strength is not None else float(saved_generation.get("lora_strength", 0) or 0)
                ),
                "reference_image": parsed.reference_image,
                "denoise_strength": parsed.denoise_strength if parsed.denoise_strength is not None else float(saved_generation.get("denoise_strength", 0) or 0),
                "prompt_weighting": bool(saved_generation.get("prompt_weighting", True)),
            }

        return build_generation_plan(
            tool,
            model_name=(Path(extra_arguments.get("base_model_path", "")).stem if base_only else model.name),
            model_path="" if base_only else model.path,
            prompt=parsed.prompt,
            image_count=count,
            steps=steps,
            seed=seed,
            sampler=sampler,
            aspect_ratio=aspect,
            extra_arguments=extra_arguments,
        )

    def _stable_diffusion_base_model_path(
        self,
        parsed: ChatGenerationRequest,
        saved_generation: dict[str, Any],
    ) -> str:
        base_assets = [
            asset
            for asset in getattr(self.planner.assets, "assets", [])
            if asset.kind == "base_model" and Path(asset.path).exists()
        ]
        if parsed.base_model_query:
            scored_bases = sorted(
                (
                    (generation_model_match_score(parsed.base_model_query, asset.name), asset)
                    for asset in base_assets
                ),
                key=lambda item: item[0],
                reverse=True,
            )
            if scored_bases and scored_bases[0][0] > 0:
                return scored_bases[0][1].path
            raise ValueError(
                f'I could not find a Stable Diffusion base model matching "{parsed.base_model_query}".'
            )
        preferred_base = next(
            (
                asset
                for asset in base_assets
                if "waiillustrious" in "".join(
                    character for character in asset.name.casefold() if character.isalnum()
                )
                or "wallilustrious" in "".join(
                    character for character in asset.name.casefold() if character.isalnum()
                )
            ),
            None,
        )
        if preferred_base is not None:
            return preferred_base.path
        selected_base = str(saved_generation.get("base_model_path", ""))
        if selected_base and Path(selected_base).expanduser().exists():
            return selected_base
        trainer_root = Path(str(self.config.get("tool_folders", {}).get("lora_trainer", "")))
        try:
            trainer_settings = json.loads(
                (trainer_root / "config" / "app_settings.json").read_text(encoding="utf-8")
            )
            configured_base = str(
                trainer_settings.get("generate_model")
                or trainer_settings.get("last_model")
                or ""
            )
            configured_path = Path(configured_base).expanduser()
            if configured_base and not configured_path.is_absolute():
                configured_path = trainer_root / configured_path
            if configured_base and configured_path.exists():
                return str(configured_path.resolve())
        except (OSError, ValueError, TypeError, json.JSONDecodeError):
            pass
        if len(base_assets) == 1:
            return base_assets[0].path
        names = ", ".join(asset.name for asset in base_assets[:4])
        available = f" Available base models: {names}." if names else ""
        raise ValueError(
            "LoRA generation also needs a Stable Diffusion base model. Put one in "
            '"LoRA StableDiffusionModels Here", or select one in the Generations tab.'
            + available
        )

    @staticmethod
    def _generation_model_is_ready(asset: Any) -> bool:
        path = Path(asset.path)
        if asset.trainer == "ddpm":
            return path.is_dir() and (path / "model_index.json").is_file()
        if asset.trainer == "flow":
            return (
                path.is_dir()
                and (path / "flow_model_info.json").is_file()
                and (path / "unet" / "config.json").is_file()
            )
        if asset.trainer == "lora":
            return (
                path.is_file()
                and path.suffix.casefold() == ".safetensors"
                and "_comfy" not in path.stem.casefold()
            ) or (
                path.is_dir()
                and any(
                    item.is_file()
                    and item.suffix.casefold() == ".safetensors"
                    and "_comfy" not in item.stem.casefold()
                    for item in path.glob("*.safetensors")
                )
            )
        return path.exists()

    def stop(self) -> None:
        if self._server is None:
            return
        self._server.shutdown()
        self._server.server_close()
        self._server = None
        self._thread = None

    def shutdown(self) -> None:
        self.stop()
        self.dispatcher.shutdown()

    def url(self) -> str:
        settings = self.settings()
        host = "127.0.0.1" if self._clean_mode(str(settings.get("remote_mode"))) == REMOTE_MODE_TAILSCALE else str(settings["bind_address"])
        return f"http://{self._url_host(host)}:{int(settings['port'])}/api/status"

    def local_test_url(self) -> str:
        settings = self.settings()
        host = "127.0.0.1" if self._clean_mode(str(settings.get("remote_mode"))) == REMOTE_MODE_TAILSCALE else str(settings["bind_address"])
        if host in {"0.0.0.0", "::"}:
            host = "127.0.0.1"
        query = urlencode({"token": str(settings["token"])})
        return f"http://{self._url_host(host)}:{int(settings['port'])}/?{query}"

    def phone_test_url(self) -> str:
        settings = self.settings()
        if self._clean_mode(str(settings.get("remote_mode"))) == REMOTE_MODE_TAILSCALE:
            return self.tailscale_url()
        host = str(settings["bind_address"])
        scope = remote_scope(host)
        if scope == "local-device only":
            return ""
        if host in {"0.0.0.0", "::"}:
            host = local_network_host()
        if not host:
            return ""
        query = urlencode({"token": str(settings["token"])})
        return f"http://{self._url_host(host)}:{int(settings['port'])}/?{query}"

    def tailscale_url(self) -> str:
        status = inspect_tailscale()
        if not status.installed or not status.connected:
            return ""
        host = status.dns_name or status.tailscale_ip
        if not host:
            return ""
        settings = self.settings()
        query = urlencode({"token": str(settings["token"])})
        return f"https://{self._url_host(host)}/?{query}"

    def tailscale_status(self) -> TailscaleStatus:
        return inspect_tailscale()

    def start_tailscale_serve(self) -> tuple[bool, str]:
        status = inspect_tailscale()
        if not status.installed:
            return False, "Tailscale is not installed."
        if not status.connected:
            return False, "Tailscale is installed but not connected."
        executable = shutil.which("tailscale")
        if not executable:
            return False, "Tailscale is not installed."
        port = int(self.settings()["port"])
        try:
            result = _run_tailscale([executable, "serve", "--bg", str(port)])
        except (OSError, subprocess.TimeoutExpired) as exc:
            return False, f"Tailscale Serve could not start: {exc}"
        if result.returncode != 0:
            return False, _command_text(result.stderr) or "Tailscale Serve could not start."
        return True, "Tailscale Serve is forwarding private tailnet traffic to ADAM."

    def stop_tailscale_serve(self) -> tuple[bool, str]:
        executable = shutil.which("tailscale")
        if not executable:
            return False, "Tailscale is not installed."
        try:
            result = _run_tailscale([executable, "serve", "reset"])
        except (OSError, subprocess.TimeoutExpired) as exc:
            return False, f"Tailscale Serve could not stop: {exc}"
        if result.returncode != 0:
            return False, _command_text(result.stderr) or "Tailscale Serve could not stop."
        return True, "Tailscale Serve forwarding was reset."

    @staticmethod
    def _url_host(host: str) -> str:
        value = host.strip() or "127.0.0.1"
        if ":" in value and not value.startswith("["):
            return f"[{value}]"
        return value

    @staticmethod
    def _clean_mode(value: str) -> str:
        mode = value.strip().casefold()
        return mode if mode in {REMOTE_MODE_DISABLED, REMOTE_MODE_LOCAL, REMOTE_MODE_TAILSCALE} else REMOTE_MODE_LOCAL