from __future__ import annotations import asyncio import base64 import json from collections.abc import Awaitable, Callable from pathlib import Path from typing import TYPE_CHECKING, Any import jinja2 from fastapi import Depends, FastAPI, HTTPException, Request from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response from device.manager import DeviceManager from host_agent.assignment import AssignmentExecutor from host_agent.client import ( HostAgentClient, HostAgentEnrollmentClient, HostAgentAPIError, HostTaskSubmissionUnknownError, ) from host_agent.config import HostAgentConfig from host_agent.conversation import ConversationAgent from host_agent.conversation_log import ConversationLogStore from host_agent.devices import register_local_device, unregister_local_device from host_agent.history import ConsoleHistoryStore from host_agent.identity import HostIdentityStore from host_agent.local_account import LocalAccountStore from host_agent.mcp_lock import McpBusyTracker from host_agent.mcp_token import McpTokenStore from host_agent.status import AgentStatusTracker from host_agent.web.auth import ( SessionManager, SessionState, attempt_login, change_password, ) from host_agent.web.mcp_auth import BearerAuthMiddleware from storage.device_config import DeviceConfigStore if TYPE_CHECKING: from mcp.server.fastmcp import FastMCP from storage.task_metadata import TaskMetadataStore from storage.timeline import Timeline SESSION_COOKIE_NAME = "host_console_session" CSRF_HEADER_NAME = "X-CSRF-Token" CSRF_FORM_FIELD = "csrf_token" _LOOPBACK_BIND_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"}) TaskSubmissionCallable = Callable[..., Awaitable[str]] TaskCancellationCallable = Callable[..., Awaitable[Any]] AUTOMATIC_DEVICE_VALUE = "__automatic__" _TERMINAL_LOCAL_TASK_STATUSES = frozenset({"completed", "failed", "cancelled"}) _ENV = jinja2.Environment( loader=jinja2.FileSystemLoader(Path(__file__).parent / "templates"), autoescape=jinja2.select_autoescape(["html", "xml"]), ) def _render( template_name: str, *, status_code: int = 200, **context: Any, ) -> HTMLResponse: html = _ENV.get_template(template_name).render(**context) return HTMLResponse(html, status_code=status_code) def _device_display_status(device: Any, *, busy_device_id: str | None) -> str: """Connected-but-idle devices report "busy" at the DeviceManager layer (an active Appium/WDA session is required to reuse it), which is not the same as a task currently running on that device. Only the device actually bound to the current assignment should read as "busy" here. """ if device.status == "busy" and device.id != busy_device_id: return "connected" return device.status def _screenshot_data_uri( record: dict[str, Any], *, path_key: str = "screenshot_path", ) -> str | None: """Return a ``data:`` URI for one step screenshot, or ``None``.""" screenshot_path = record.get(path_key) if not screenshot_path: return None path = Path(str(screenshot_path)) if not path.exists(): return None encoded = base64.b64encode(path.read_bytes()).decode("ascii") return f"data:image/png;base64,{encoded}" def _ocr_results(record: dict[str, Any]) -> list[dict[str, Any]]: raw_results = record.get("ocr_results") if not isinstance(raw_results, list): return [] return [result for result in raw_results if isinstance(result, dict)] def _ui_tree_nodes(record: dict[str, Any]) -> list[dict[str, Any]]: raw_nodes = record.get("ui_tree_results") if not isinstance(raw_nodes, list): return [] return [node for node in raw_nodes if isinstance(node, dict)] def _overlay_payload(record: dict[str, Any]) -> dict[str, Any]: """Combined perception elements + screen size for client-side bounding-box overlay and action-effect rendering on the before-screenshot. """ scene = record.get("scene") screen = scene.get("screen") if isinstance(scene, dict) else None width = screen.get("width") if isinstance(screen, dict) else None height = screen.get("height") if isinstance(screen, dict) else None elements = scene.get("elements") if isinstance(scene, dict) else None return { "width": width if isinstance(width, (int, float)) else 0, "height": height if isinstance(height, (int, float)) else 0, "elements": [element for element in elements if isinstance(element, dict)] if isinstance(elements, list) else [], } def _timeline_step_context(record: dict[str, Any]) -> dict[str, Any]: tool_call = record.get("tool_call") result = record.get("result") return { "index": record.get("index", ""), "timestamp": record.get("timestamp", ""), "prompt": record.get("prompt") or "", "tool_call": tool_call if isinstance(tool_call, dict) else {}, "result": result if isinstance(result, dict) else {}, "before_screenshot_src": _screenshot_data_uri( record, path_key="before_screenshot_path" ), "after_screenshot_src": _screenshot_data_uri( record, path_key="after_screenshot_path" ) or _screenshot_data_uri(record), "ocr_results": _ocr_results(record), "ui_tree_nodes": _ui_tree_nodes(record), "overlay": _overlay_payload(record), } def _safe_submission_error(detail: str) -> str: """Return a safe, single-line error message for the operator. The Cloud response ``detail`` is treated as a static control-plane message; we strip surrounding whitespace and reject empty results so the operator never sees a blank error or, through Jinja autoescape, anything that could carry unrendered HTML. """ cleaned = " ".join(str(detail).split()).strip() return cleaned or "Cloud rejected the task submission." def _extract_task_id(response: Any) -> str | None: """Normalize the Host self-submission return value to a Cloud task ID. Tests and narrow protocol overrides may return a bare string while the production client returns a pydantic model. Accept either so the rest of the handler can rely on a single string ID. """ candidate: Any = response if hasattr(candidate, "task_id"): candidate = getattr(candidate, "task_id") if not isinstance(candidate, str): return None cleaned = candidate.strip() return cleaned or None def _dashboard_texts(*, snapshot: dict[str, Any]) -> dict[str, str]: """Pre-compute human-readable text strings for the dashboard template.""" heartbeat = snapshot.get("last_heartbeat") assignment = snapshot.get("current_assignment") policy = snapshot.get("host_policy") progress = snapshot.get("progress") return { "heartbeat_text": ( f"{'ok' if heartbeat['ok'] else 'failed'} at {heartbeat['at']} " f"({heartbeat['device_count']} devices)" if heartbeat else "never" ), "assignment_text": ( f"{assignment['task_id']} on {assignment['device_id']} " f"(started {assignment['started_at']})" if assignment else "none" ), "progress_text": ( f"step {progress['step_index']} \u2014 {progress['step_status']}: " f"{progress['summary']}" if progress else "" ), "policy_text": ( f"revision {policy['revision']}; self-submission " f"{'enabled' if policy['self_submission_enabled'] else 'disabled'}; " f"max active tasks {policy['max_active_tasks'] or 'unlimited'}; " f"daily token budget {policy['daily_token_budget'] or 'unmetered'}" if policy else "no Cloud policy cached" ), } def create_console_app( *, config: HostAgentConfig, manager: DeviceManager, config_store: DeviceConfigStore, local_account_store: LocalAccountStore, identity_store: HostIdentityStore, history_store: ConsoleHistoryStore, status_tracker: AgentStatusTracker, session_manager: SessionManager, enrollment_client: HostAgentEnrollmentClient | None, host_client: HostAgentClient | None = None, submit_self_task: TaskSubmissionCallable | None = None, cancel_task: TaskCancellationCallable | None = None, metadata_store: TaskMetadataStore | None = None, timeline: Timeline | None = None, executor: AssignmentExecutor | None = None, mcp_server: FastMCP | None = None, mcp_token_store: McpTokenStore | None = None, mcp_busy_tracker: McpBusyTracker | None = None, conversation_agent: ConversationAgent | None = None, conversation_log: ConversationLogStore | None = None, ) -> FastAPI: app = FastAPI(title="Host Agent Console") cookie_secure = config.console_bind_host not in _LOOPBACK_BIND_HOSTS if submit_self_task is None and host_client is not None: submit_self_task = host_client.submit_self_task if cancel_task is None and host_client is not None: cancel_task = host_client.cancel_task submission_available = submit_self_task is not None mcp_mounted = mcp_server is not None and mcp_token_store is not None if mcp_mounted: from starlette.applications import Starlette from starlette.middleware import Middleware mcp_asgi = mcp_server.streamable_http_app() authed = Starlette( routes=[], middleware=[Middleware(BearerAuthMiddleware, token_store=mcp_token_store)], ) authed.router.mount("/", mcp_asgi) app.mount("/mcp", authed) def _running_devices() -> list[dict[str, str]]: return [ { "id": device.id, "label": device.name or device.id, } for device in manager.list_devices() ] def _session_token(request: Request) -> str | None: return request.cookies.get(SESSION_COOKIE_NAME) def require_session(request: Request) -> SessionState: token = _session_token(request) state = session_manager.validate(token) if token else None if state is None: raise HTTPException(status_code=303, headers={"Location": "/login"}) return state async def require_csrf( request: Request, session: SessionState = Depends(require_session), ) -> SessionState: provided = request.headers.get(CSRF_HEADER_NAME) if not provided: form = await request.form() raw = form.get(CSRF_FORM_FIELD) provided = str(raw) if raw is not None else None token = _session_token(request) if ( not token or not provided or not session_manager.validate_csrf(token, provided) ): raise HTTPException(status_code=403, detail="invalid CSRF token") return session @app.get("/login", response_class=HTMLResponse) async def login_page() -> HTMLResponse: account = await asyncio.to_thread(local_account_store.load) return _render("login.html", title="Login", session=None, account=account) @app.post("/login") async def login_submit(request: Request) -> Response: account = await asyncio.to_thread(local_account_store.load) if account is None: return _render("login.html", title="Login", session=None, account=None) form = await request.form() username = str(form.get("username", "")) password = str(form.get("password", "")) ok = await asyncio.to_thread( attempt_login, local_account_store, username=username, password=password ) if not ok: return _render( "login.html", title="Login", session=None, account=account, error="Invalid username or password.", ) session_token, _ = session_manager.create_session(username) response = RedirectResponse(url="/", status_code=303) response.set_cookie( key=SESSION_COOKIE_NAME, value=session_token, httponly=True, samesite="strict", secure=cookie_secure, path="/", ) return response @app.post("/logout") async def logout( request: Request, session: SessionState = Depends(require_csrf), ) -> Response: token = _session_token(request) if token: session_manager.invalidate(token) response = RedirectResponse(url="/login", status_code=303) response.delete_cookie(key=SESSION_COOKIE_NAME, path="/") return response @app.get("/", response_class=HTMLResponse) async def dashboard( session: SessionState = Depends(require_session), ) -> HTMLResponse: identity = await asyncio.to_thread(identity_store.load) snapshot = status_tracker.snapshot() busy_device_id = ( snapshot["current_assignment"]["device_id"] if snapshot.get("current_assignment") else None ) devices = [ { "id": d.id, "name": d.name, "driver_type": d.driver_type, "display_status": _device_display_status( d, busy_device_id=busy_device_id ), } for d in manager.list_devices() ] texts = _dashboard_texts(snapshot=snapshot) mcp_endpoint = "/mcp" if mcp_mounted else None mcp_busy_devices = ( mcp_busy_tracker.busy_device_ids() if mcp_busy_tracker is not None else [] ) return _render( "dashboard.html", title="Status", session=session, identity=identity, devices=devices, config=config, mcp_endpoint=mcp_endpoint, mcp_busy_devices=mcp_busy_devices, **texts, ) @app.get("/api/status") async def api_status( session: SessionState = Depends(require_session), ) -> JSONResponse: snapshot = status_tracker.snapshot() if executor is not None and snapshot.get("progress") is None: live = executor.latest_progress() if live is not None: snapshot["progress"] = { "step_index": live.step_index, "step_status": live.step_status, "summary": live.summary, "updated_at": live.updated_at.isoformat(), } current_assignment = snapshot.get("current_assignment") busy_device_id = current_assignment["device_id"] if current_assignment else None devices = [ { "id": device.id, "name": device.name, "driver_type": device.driver_type, "status": _device_display_status(device, busy_device_id=busy_device_id), } for device in manager.list_devices() ] return JSONResponse( { "status": snapshot, "devices": devices, "mcp_endpoint": "/mcp" if mcp_mounted else None, "mcp_busy_devices": ( mcp_busy_tracker.busy_device_ids() if mcp_busy_tracker is not None else [] ), } ) @app.post("/api/chat") async def api_chat( request: Request, session: SessionState = Depends(require_csrf), ) -> JSONResponse: if conversation_agent is None: raise HTTPException(status_code=503, detail="chat agent is not configured") payload = await request.json() device_id = payload.get("device_id") if isinstance(payload, dict) else None if not isinstance(device_id, str) or not device_id.strip(): raise HTTPException(status_code=400, detail="device_id is required") if device_id not in {device.id for device in manager.list_devices()}: raise HTTPException(status_code=404, detail="unknown device") raw_messages = payload.get("messages") if isinstance(payload, dict) else None if not isinstance(raw_messages, list) or not raw_messages: raise HTTPException(status_code=400, detail="messages must be a non-empty list") messages = [ {key: value for key, value in item.items() if key in {"role", "content", "image_base64", "mime_type", "text"}} for item in raw_messages if isinstance(item, dict) and item.get("role") in {"user", "assistant"} and (isinstance(item.get("content"), (str, list)) or isinstance(item.get("image_base64"), str)) ] if not messages: raise HTTPException(status_code=400, detail="messages are invalid") result = await asyncio.to_thread( conversation_agent.chat_for_device, device_id.strip(), messages ) return JSONResponse( {"content": result.content, "tool_calls": result.tool_calls} ) @app.get("/devices", response_class=HTMLResponse) async def devices_page( request: Request, session: SessionState = Depends(require_session), ) -> HTMLResponse: devices = await asyncio.to_thread(config_store.list) edit_id = request.query_params.get("edit") edit_record = ( await asyncio.to_thread(config_store.get, edit_id) if edit_id else None ) connection_info_json = ( json.dumps(edit_record["connection_info"]) if edit_record else "{}" ) return _render( "devices.html", title="Devices", session=session, devices=devices, csrf_token=session.csrf_token, edit_record=edit_record, connection_info_json=connection_info_json, error=None, ) @app.post("/devices/save") async def devices_save( request: Request, session: SessionState = Depends(require_csrf), ) -> Response: form = await request.form() device_id = str(form.get("device_id", "")).strip() driver_type = str(form.get("driver_type", "")).strip() name = str(form.get("name", "")).strip() or None connection_info_raw = str(form.get("connection_info", "") or "{}") error: str | None = None connection_info: dict[str, Any] = {} if not device_id or not driver_type: error = "Device ID and driver type are required." else: try: parsed = json.loads(connection_info_raw) except ValueError: error = "Connection info must be valid JSON." else: if not isinstance(parsed, dict): error = "Connection info must be a JSON object." else: connection_info = parsed if error is None: try: await asyncio.to_thread( register_local_device, config_store, manager, device_id=device_id, driver_type=driver_type, connection_info=connection_info, name=name, config=config, enrollment_client=enrollment_client, ) except ValueError as exc: error = str(exc) if error is not None: devices = await asyncio.to_thread(config_store.list) return _render( "devices.html", title="Devices", session=session, devices=devices, csrf_token=session.csrf_token, edit_record=None, connection_info_json="{}", error=error, status_code=400, ) return RedirectResponse(url="/devices", status_code=303) @app.post("/devices/remove") async def devices_remove( request: Request, session: SessionState = Depends(require_csrf), ) -> Response: form = await request.form() device_id = str(form.get("device_id", "")).strip() if device_id: await asyncio.to_thread( unregister_local_device, config_store, manager, device_id=device_id ) return RedirectResponse(url="/devices", status_code=303) @app.get("/account", response_class=HTMLResponse) async def account_page( session: SessionState = Depends(require_session), ) -> HTMLResponse: return _render( "account.html", title="Account", session=session, csrf_token=session.csrf_token, message=None, error=None, ) @app.post("/account", response_class=HTMLResponse) async def account_submit( request: Request, session: SessionState = Depends(require_csrf), ) -> HTMLResponse: form = await request.form() current_password = str(form.get("current_password", "")) new_password = str(form.get("new_password", "")) confirm_password = str(form.get("confirm_password", "")) if not new_password or new_password != confirm_password: return _render( "account.html", title="Account", session=session, csrf_token=session.csrf_token, message=None, error="New password and confirmation must match.", status_code=400, ) ok = await asyncio.to_thread( change_password, local_account_store, current_password=current_password, new_password=new_password, ) if not ok: return _render( "account.html", title="Account", session=session, csrf_token=session.csrf_token, message=None, error="Current password is incorrect.", status_code=400, ) return _render( "account.html", title="Account", session=session, csrf_token=session.csrf_token, message="Password updated.", error=None, ) @app.get("/history", response_class=HTMLResponse) async def history_page( session: SessionState = Depends(require_session), ) -> HTMLResponse: entries = await asyncio.to_thread(history_store.list_recent) return _render( "history.html", title="History", session=session, entries=entries, ) @app.get("/conversations", response_class=HTMLResponse) async def conversations_page( session: SessionState = Depends(require_session), ) -> HTMLResponse: events = await asyncio.to_thread( conversation_log.list_recent if conversation_log is not None else (lambda: []) ) return _render("conversations.html", title="Conversations", session=session, events=events) def _tasks_list_context( session: SessionState, *, goal_value: str = "", selected_device: str = AUTOMATIC_DEVICE_VALUE, submission_error: str | None = None, submission_notice: str | None = None, submission_unknown: bool = False, status_code: int = 200, ) -> dict[str, Any]: devices = _running_devices() return { "title": "Tasks", "session": session, "csrf_token": session.csrf_token, "tasks": metadata_store.list_tasks() if metadata_store is not None else [], "metadata_store_missing": metadata_store is None, "devices": devices, "automatic_device_value": AUTOMATIC_DEVICE_VALUE, "selected_device": selected_device if any(d["id"] == selected_device for d in devices) or selected_device == AUTOMATIC_DEVICE_VALUE else AUTOMATIC_DEVICE_VALUE, "submission_available": submission_available, "goal_value": goal_value, "submission_error": submission_error, "submission_notice": submission_notice, "submission_unknown": submission_unknown, "status_code": status_code, } @app.get("/tasks", response_class=HTMLResponse) async def tasks_page( request: Request, session: SessionState = Depends(require_session), ) -> HTMLResponse: if metadata_store is None: raise HTTPException( status_code=503, detail="task metadata store not configured" ) notice: str | None = None unknown = False error: str | None = None if request.query_params.get("submitted") == "1": task_id = request.query_params.get("task_id", "") if task_id: notice = ( f"Task submitted. Cloud task ID: {task_id}. " "It will appear below when this Host begins executing it." ) elif request.query_params.get("outcome") == "unknown": unknown = True context = _tasks_list_context( session, submission_notice=notice, submission_unknown=unknown, submission_error=error, ) return _render( "tasks_list.html", status_code=context["status_code"], **{k: v for k, v in context.items() if k != "status_code"}, ) @app.post("/tasks/submit", response_class=HTMLResponse) async def tasks_submit( request: Request, session: SessionState = Depends(require_csrf), ) -> Response: if metadata_store is None: raise HTTPException( status_code=503, detail="task metadata store not configured" ) if submit_self_task is None: context = _tasks_list_context( session, submission_error=( "Host submission client is not available yet. " "Wait for Host enrollment to complete, then retry." ), ) return _render( "tasks_list.html", status_code=503, **{k: v for k, v in context.items() if k != "status_code"}, ) form = await request.form() raw_goal = str(form.get("goal", "")) goal = raw_goal.strip() device_selection = str(form.get("device_id", AUTOMATIC_DEVICE_VALUE)).strip() explicit_device_id: str | None = None if device_selection and device_selection != AUTOMATIC_DEVICE_VALUE: snapshot_ids = {device.id for device in manager.list_devices()} if device_selection not in snapshot_ids: context = _tasks_list_context( session, goal_value=goal, selected_device=device_selection, submission_error=( "Selected device is no longer registered. " "Refresh and try again." ), ) return _render( "tasks_list.html", status_code=400, **{k: v for k, v in context.items() if k != "status_code"}, ) explicit_device_id = device_selection if not goal: context = _tasks_list_context( session, goal_value=goal, selected_device=device_selection or AUTOMATIC_DEVICE_VALUE, submission_error="Goal cannot be empty.", ) return _render( "tasks_list.html", status_code=400, **{k: v for k, v in context.items() if k != "status_code"}, ) try: response = await submit_self_task(goal=goal, device_id=explicit_device_id) except HostAgentAPIError as exc: context = _tasks_list_context( session, goal_value=goal, selected_device=device_selection, submission_error=_safe_submission_error(str(exc.detail)), ) return _render( "tasks_list.html", status_code=502, **{k: v for k, v in context.items() if k != "status_code"}, ) except HostTaskSubmissionUnknownError: return RedirectResponse( url="/tasks?outcome=unknown", status_code=303, ) task_id = _extract_task_id(response) if task_id is None: context = _tasks_list_context( session, goal_value=goal, selected_device=device_selection, submission_error=( "Host submission client returned an unexpected response." ), ) return _render( "tasks_list.html", status_code=502, **{k: v for k, v in context.items() if k != "status_code"}, ) try: await asyncio.to_thread( history_store.record_task_submission, task_id=task_id, device_id=explicit_device_id, ) except Exception: pass params = f"submitted=1&task_id={task_id}" return RedirectResponse(url=f"/tasks?{params}", status_code=303) @app.get("/tasks/{task_id}", response_class=HTMLResponse) async def task_detail_page( task_id: str, request: Request, session: SessionState = Depends(require_session), ) -> HTMLResponse: if metadata_store is None: raise HTTPException( status_code=503, detail="task metadata store not configured" ) task = await asyncio.to_thread(metadata_store.get_task, task_id) if task is None: raise HTTPException(status_code=404, detail="task not found") timeline_records: list[dict[str, Any]] = [] if timeline is not None: timeline_records = await asyncio.to_thread(timeline.read, task_id) task_rows = [ (label, task[key]) for key, label in ( ("source_task_id", "Cloud task ID"), ("source_attempt", "Cloud attempt"), ("id", "Execution ID"), ("goal", "Goal"), ("device_id", "Device"), ("status", "Status"), ("created_at", "Created"), ("updated_at", "Updated"), ("completed_at", "Completed"), ("failure_reason", "Failure reason"), ) if task.get(key) is not None ] timeline_steps = [_timeline_step_context(record) for record in timeline_records] can_cancel = ( cancel_task is not None and task.get("source_task_id") is not None and task.get("status") not in _TERMINAL_LOCAL_TASK_STATUSES ) cancel_notice = ( "Cancellation requested. It may take a moment to take effect." if request.query_params.get("cancelled") == "1" else None ) cancel_error = ( "Failed to request cancellation. Try again." if request.query_params.get("cancel_error") == "1" else None ) return _render( "task_detail.html", title=f"Task {task_id}", session=session, csrf_token=session.csrf_token, task=task, task_rows=task_rows, timeline_steps=timeline_steps, can_cancel=can_cancel, cancel_notice=cancel_notice, cancel_error=cancel_error, ) @app.post("/tasks/{task_id}/cancel") async def tasks_cancel( task_id: str, session: SessionState = Depends(require_csrf), ) -> Response: if metadata_store is None: raise HTTPException( status_code=503, detail="task metadata store not configured" ) task = await asyncio.to_thread(metadata_store.get_task, task_id) if task is None: raise HTTPException(status_code=404, detail="task not found") source_task_id = task.get("source_task_id") if cancel_task is None or source_task_id is None: return RedirectResponse( url=f"/tasks/{task_id}?cancel_error=1", status_code=303 ) try: await cancel_task(source_task_id) except HostAgentAPIError: return RedirectResponse( url=f"/tasks/{task_id}?cancel_error=1", status_code=303 ) return RedirectResponse(url=f"/tasks/{task_id}?cancelled=1", status_code=303) return app