agentic-mobile-control/apps/device-host-agent/host_agent/web/app.py

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 core.errors import DeviceNotFoundError, DeviceOfflineError, DeviceRuntimeError
from device.manager import DeviceManager
from driver.registry import build_driver_factory
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.ios_discovery import IOSDiscoveryError, discover_connected_ios_devices
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")
        if conversation_log is not None:
            await asyncio.to_thread(
                conversation_log.append,
                {"type": "user_request", "device_id": device_id.strip(), "messages": messages},
            )
        try:
            result = await asyncio.to_thread(
                conversation_agent.chat_for_device, device_id.strip(), messages
            )
        except Exception as exc:
            if conversation_log is not None:
                await asyncio.to_thread(
                    conversation_log.append,
                    {"type": "agent_error", "device_id": device_id.strip(), "error": str(exc)},
                )
            raise HTTPException(status_code=502, detail=str(exc)) from exc
        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("/api/devices/{device_id}/screenshot")
    async def api_device_screenshot(
        device_id: str,
        session: SessionState = Depends(require_csrf),
    ) -> Response:
        """Capture one on-demand screenshot for a connected local device."""
        try:
            screenshot = await asyncio.to_thread(
                lambda: manager.active_driver(device_id).screenshot()
            )
        except DeviceNotFoundError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        except DeviceOfflineError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
        except DeviceRuntimeError as exc:
            raise HTTPException(status_code=502, detail=str(exc)) from exc
        except Exception as exc:
            raise HTTPException(
                status_code=502,
                detail=str(exc) or "failed to capture device screenshot",
            ) from exc

        if not isinstance(screenshot, bytes) or not screenshot:
            raise HTTPException(
                status_code=502,
                detail="device returned an empty screenshot",
            )
        return Response(
            content=screenshot,
            media_type="image/png",
            headers={
                "Cache-Control": "no-store",
                "Pragma": "no-cache",
                "X-Content-Type-Options": "nosniff",
            },
        )

    @app.get("/api/devices/discover-ios")
    async def api_discover_ios_devices(
        session: SessionState = Depends(require_session),
    ) -> JSONResponse:
        try:
            discovered = await asyncio.to_thread(discover_connected_ios_devices)
        except IOSDiscoveryError as exc:
            raise HTTPException(status_code=502, detail=str(exc)) from exc

        configured = await asyncio.to_thread(config_store.list)
        configured_udids = {
            str(record["connection_info"].get("udid"))
            for record in configured
            if record["driver_type"] == "wda"
        }
        used_wda_ports = {
            record["connection_info"].get("wda_local_port") for record in configured
        }
        used_mjpeg_ports = {
            record["connection_info"].get("mjpegServerPort") for record in configured
        }
        next_wda_port = 8100
        next_mjpeg_port = 9100
        result = []
        for device in discovered:
            while next_wda_port in used_wda_ports:
                next_wda_port += 1
            while next_mjpeg_port in used_mjpeg_ports:
                next_mjpeg_port += 1
            result.append(
                {
                    **device,
                    "configured": device["udid"] in configured_udids,
                    "suggested_wda_port": next_wda_port,
                    "suggested_mjpeg_port": next_mjpeg_port,
                }
            )
            used_wda_ports.add(next_wda_port)
            used_mjpeg_ports.add(next_mjpeg_port)
            next_wda_port += 1
            next_mjpeg_port += 1
        return JSONResponse({"devices": result})

    @app.post("/api/devices/test-connection")
    async def api_device_test_connection(
        request: Request,
        session: SessionState = Depends(require_csrf),
    ) -> JSONResponse:
        payload = await request.json()
        if not isinstance(payload, dict):
            raise HTTPException(status_code=400, detail="Request must be an object.")
        driver_type = str(payload.get("driver_type", "")).strip()
        connection_info = payload.get("connection_info")
        if not driver_type or not isinstance(connection_info, dict):
            raise HTTPException(
                status_code=400,
                detail="Driver type and connection info are required.",
            )

        driver = None
        connected = False
        try:
            driver = build_driver_factory(driver_type, connection_info)()
            await asyncio.to_thread(driver.connect)
            connected = True
            await asyncio.to_thread(driver.health_check)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        except Exception as exc:
            raise HTTPException(
                status_code=502,
                detail=str(exc) or "device connection test failed",
            ) from exc
        finally:
            if connected and driver is not None:
                try:
                    await asyncio.to_thread(driver.disconnect)
                except Exception:
                    pass
        return JSONResponse({"ok": True, "driver_type": driver_type})

    @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:
            # The console exposes the common Appium settings as regular form
            # fields. Advanced JSON remains available for uncommon capabilities.
            field_map = {
                "server_url": "server_url",
                "udid": "udid",
                "device_name": "device_name",
            }
            for form_key, config_key in field_map.items():
                value = str(form.get(form_key, "")).strip()
                if value:
                    connection_info[config_key] = value
            port_key = "wda_local_port" if driver_type == "wda" else "system_port"
            port_value = str(form.get(port_key, "")).strip()
            if port_value:
                try:
                    connection_info[port_key] = int(port_value)
                except ValueError:
                    error = f"{port_key} must be an integer."

        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