import asyncio
import base64
import hashlib
import json
import os
import uuid
from collections import OrderedDict
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Literal

from playwright.async_api import (
    BrowserContext,
    Page,
    Playwright,
    async_playwright,
)
from playwright.async_api import (
    Error as PlaywrightError,
)
from playwright.async_api import (
    TimeoutError as PlaywrightTimeoutError,
)

from .config import Settings
from .errors import ComputerRuntimeError
from .network import validate_public_url
from .schemas import (
    ActionResult,
    BrowserActionInput,
    HumanKeyboardInput,
    HumanPointerInput,
    SessionState,
    StartSessionInput,
)

ACTIVE_RUNTIME_STATES = {"ready", "ai_controlled", "human_controlled", "paused"}
KEY_PATTERN = frozenset({
    "Enter", "Tab", "Escape", "Backspace", "Delete", "ArrowUp", "ArrowDown",
    "ArrowLeft", "ArrowRight", "Home", "End", "PageUp", "PageDown", "Space",
})


@dataclass(slots=True)
class RuntimeSession:
    session_id: uuid.UUID
    workspace_id: uuid.UUID
    profile_id: uuid.UUID
    profile_storage_key: str
    profile_path: Path
    worker_id: str
    fencing_token: int
    status: Literal["ready", "ai_controlled", "human_controlled", "paused", "stopped"]
    context: BrowserContext | None
    tab_ids: dict[Page, str] = field(default_factory=dict)
    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
    recent_requests: OrderedDict[str, str] = field(default_factory=OrderedDict)


class ComputerRuntime:
    def __init__(self, settings: Settings):
        self.settings = settings
        self.profile_root = Path(settings.browser_profile_root).resolve()
        self.artifact_root = Path(settings.computer_artifact_root).resolve()
        self.playwright: Playwright | None = None
        self.sessions: dict[uuid.UUID, RuntimeSession] = {}
        self.profile_owners: dict[tuple[uuid.UUID, uuid.UUID], uuid.UUID] = {}
        self._manager_lock = asyncio.Lock()

    async def startup(self) -> None:
        self.profile_root.mkdir(parents=True, exist_ok=True)
        self.artifact_root.mkdir(parents=True, exist_ok=True)
        self.playwright = await async_playwright().start()

    async def shutdown(self) -> None:
        for runtime in list(self.sessions.values()):
            if runtime.context:
                try:
                    await runtime.context.close()
                except PlaywrightError:
                    pass
        self.sessions.clear()
        self.profile_owners.clear()
        if self.playwright:
            await self.playwright.stop()
            self.playwright = None

    def _profile_path(self, payload: StartSessionInput) -> Path:
        candidate = (
            self.profile_root
            / str(payload.workspace_id)
            / str(payload.profile_id)
            / payload.profile_storage_key
        ).resolve()
        if not candidate.is_relative_to(self.profile_root):
            raise ComputerRuntimeError(
                "BROWSER_PROFILE_PATH_INVALID", "The browser profile path is invalid.", 422
            )
        return candidate

    @staticmethod
    def _manifest_path(runtime: RuntimeSession) -> Path:
        return runtime.profile_path / "hayva-runtime-session.json"

    def _write_manifest(self, runtime: RuntimeSession) -> None:
        manifest = self._manifest_path(runtime)
        temporary = manifest.with_suffix(".tmp")
        payload = {
            "schema_version": 1,
            "session_id": str(runtime.session_id),
            "workspace_id": str(runtime.workspace_id),
            "profile_id": str(runtime.profile_id),
            "worker_id": runtime.worker_id,
            "fencing_token": runtime.fencing_token,
            "status": runtime.status,
            "updated_at": datetime.now(UTC).isoformat(),
        }
        temporary.write_text(
            json.dumps(payload, sort_keys=True, separators=(",", ":")), encoding="utf-8"
        )
        os.replace(temporary, manifest)

    def _read_manifest(self, profile_path: Path) -> dict[str, Any] | None:
        manifest = profile_path / "hayva-runtime-session.json"
        try:
            value = json.loads(manifest.read_text(encoding="utf-8"))
        except (OSError, ValueError, TypeError):
            return None
        return value if isinstance(value, dict) and value.get("schema_version") == 1 else None

    async def _guard_route(self, route) -> None:
        try:
            await validate_public_url(route.request.url, allow_internal_browser_url=True)
        except ComputerRuntimeError:
            await route.abort("blockedbyclient")
            return
        await route.continue_()

    async def _launch_context(self, profile_path: Path) -> BrowserContext:
        if not self.playwright:
            raise ComputerRuntimeError(
                "BROWSER_RUNTIME_UNAVAILABLE", "The browser runtime is unavailable.", 503
            )
        context = await self.playwright.chromium.launch_persistent_context(
            str(profile_path),
            headless=self.settings.browser_headless,
            accept_downloads=False,
            service_workers="block",
            viewport={"width": 1440, "height": 900},
            args=["--disable-dev-shm-usage"],
        )
        await context.route("**/*", self._guard_route)
        # WebSockets are denied until a DNS-rebinding-safe socket guard is implemented.
        if hasattr(context, "route_web_socket"):
            await context.route_web_socket("**/*", lambda socket_route: socket_route.close())
        context.set_default_timeout(self.settings.browser_action_timeout_ms)
        context.set_default_navigation_timeout(self.settings.browser_action_timeout_ms)
        return context

    def _register_pages(self, runtime: RuntimeSession) -> None:
        if not runtime.context:
            return
        for page in runtime.context.pages:
            runtime.tab_ids.setdefault(page, uuid.uuid4().hex)

        def register(page: Page) -> None:
            runtime.tab_ids[page] = uuid.uuid4().hex

        runtime.context.on("page", register)

    async def start_session(
        self, payload: StartSessionInput, *, request_id: str
    ) -> SessionState:
        async with self._manager_lock:
            existing = self.sessions.get(payload.session_id)
            if existing:
                async with existing.lock:
                    self._accept_fence(existing, payload.fencing_token)
                    if existing.context is None or existing.status == "stopped":
                        raise ComputerRuntimeError(
                            "BROWSER_SESSION_STOPPED", "The browser session is stopped.", 409
                        )
                    return await self._state(existing)
            owner_key = (payload.workspace_id, payload.profile_id)
            owner = self.profile_owners.get(owner_key)
            if owner and owner != payload.session_id:
                raise ComputerRuntimeError(
                    "BROWSER_PROFILE_BUSY", "The browser profile is already in use.", 409
                )
            active_count = sum(
                item.context is not None and item.status != "stopped"
                for item in self.sessions.values()
            )
            if active_count >= self.settings.browser_max_sessions:
                raise ComputerRuntimeError(
                    "BROWSER_CAPACITY_REACHED", "Browser session capacity is reached.", 503
                )
            profile_path = self._profile_path(payload)
            profile_path.mkdir(parents=True, exist_ok=True)
            manifest = self._read_manifest(profile_path)
            if manifest and manifest.get("session_id") not in {
                str(payload.session_id), None
            } and manifest.get("status") != "stopped":
                raise ComputerRuntimeError(
                    "BROWSER_PROFILE_RECOVERY_CONFLICT",
                    "The browser profile belongs to an unfinished session.",
                    409,
                )
            previous_token = 0
            if manifest and manifest.get("session_id") == str(payload.session_id):
                try:
                    previous_token = int(manifest.get("fencing_token", 0))
                except (TypeError, ValueError):
                    previous_token = 0
                if payload.fencing_token < previous_token:
                    raise ComputerRuntimeError(
                        "BROWSER_FENCING_TOKEN_STALE", "The control lease is stale.", 409
                    )
            try:
                context = await self._launch_context(profile_path)
            except PlaywrightError as error:
                raise ComputerRuntimeError(
                    "BROWSER_LAUNCH_FAILED", "The isolated browser could not start.", 503
                ) from error
            runtime = RuntimeSession(
                session_id=payload.session_id,
                workspace_id=payload.workspace_id,
                profile_id=payload.profile_id,
                profile_storage_key=payload.profile_storage_key,
                profile_path=profile_path,
                worker_id=f"computer-{uuid.uuid4().hex}",
                fencing_token=max(payload.fencing_token, previous_token),
                status="ai_controlled",
                context=context,
            )
            self._register_pages(runtime)
            self._remember_request(runtime, request_id, "start")
            self.sessions[runtime.session_id] = runtime
            self.profile_owners[owner_key] = runtime.session_id
            self._write_manifest(runtime)
            return await self._state(runtime)

    @staticmethod
    def _accept_fence(runtime: RuntimeSession, fencing_token: int) -> None:
        if fencing_token < runtime.fencing_token:
            raise ComputerRuntimeError(
                "BROWSER_FENCING_TOKEN_STALE", "The control lease is stale.", 409
            )
        runtime.fencing_token = fencing_token

    @staticmethod
    def _remember_request(runtime: RuntimeSession, request_id: str, operation: str) -> None:
        prior = runtime.recent_requests.get(request_id)
        if prior and prior != operation:
            raise ComputerRuntimeError(
                "BROWSER_IDEMPOTENCY_CONFLICT", "The request identifier was reused.", 409
            )
        runtime.recent_requests[request_id] = operation
        runtime.recent_requests.move_to_end(request_id)
        while len(runtime.recent_requests) > 200:
            runtime.recent_requests.popitem(last=False)

    def _get_runtime(self, session_id: uuid.UUID) -> RuntimeSession:
        runtime = self.sessions.get(session_id)
        if not runtime:
            raise ComputerRuntimeError(
                "BROWSER_SESSION_NOT_FOUND", "The browser session is unavailable.", 404
            )
        return runtime

    async def command(
        self,
        session_id: uuid.UUID,
        command: Literal["pause", "resume", "stop"],
        fencing_token: int,
        *,
        request_id: str,
    ) -> SessionState:
        runtime = self._get_runtime(session_id)
        async with runtime.lock:
            self._accept_fence(runtime, fencing_token)
            self._remember_request(runtime, request_id, command)
            if command == "pause":
                if runtime.status != "stopped":
                    runtime.status = "paused"
            elif command == "resume":
                if runtime.context is None or runtime.status == "stopped":
                    raise ComputerRuntimeError(
                        "BROWSER_SESSION_STOPPED", "The browser session is stopped.", 409
                    )
                runtime.status = "ai_controlled"
            else:
                if runtime.context:
                    await runtime.context.close()
                    runtime.context = None
                runtime.status = "stopped"
                self.profile_owners.pop((runtime.workspace_id, runtime.profile_id), None)
            self._write_manifest(runtime)
            return await self._state(runtime)

    async def snapshot(
        self, session_id: uuid.UUID, fencing_token: int, *, request_id: str
    ) -> SessionState:
        runtime = self._get_runtime(session_id)
        async with runtime.lock:
            self._accept_fence(runtime, fencing_token)
            self._remember_request(runtime, request_id, "snapshot")
            if runtime.status == "paused":
                runtime.status = "human_controlled"
            self._write_manifest(runtime)
            return await self._state(runtime)

    async def _page(self, runtime: RuntimeSession) -> Page:
        if runtime.context is None or runtime.status == "stopped":
            raise ComputerRuntimeError(
                "BROWSER_SESSION_STOPPED", "The browser session is stopped.", 409
            )
        pages = [page for page in runtime.context.pages if not page.is_closed()]
        if not pages:
            page = await runtime.context.new_page()
            runtime.tab_ids[page] = uuid.uuid4().hex
            return page
        return pages[-1]

    async def _state(self, runtime: RuntimeSession) -> SessionState:
        if runtime.context is None:
            return SessionState(
                worker_id=runtime.worker_id, status="stopped",
                current_url=None, active_tab_id=None, tabs=[], page_title=None,
                observed_at=datetime.now(UTC).isoformat(),
            )
        pages = [page for page in runtime.context.pages if not page.is_closed()]
        tabs = []
        for page in pages[:50]:
            tab_id = runtime.tab_ids.setdefault(page, uuid.uuid4().hex)
            try:
                title = (await page.title())[:500]
            except PlaywrightError:
                title = ""
            tabs.append({"id": tab_id, "url": page.url[:4096], "title": title})
        active = pages[-1] if pages else None
        try:
            active_title = (await active.title())[:500] if active else None
        except PlaywrightError:
            active_title = None
        return SessionState(
            worker_id=runtime.worker_id,
            status=runtime.status,
            current_url=active.url[:4096] if active else None,
            active_tab_id=runtime.tab_ids.get(active) if active else None,
            tabs=tabs,
            page_title=active_title,
            observed_at=datetime.now(UTC).isoformat(),
        )

    @staticmethod
    def _argument(arguments: dict[str, Any], name: str, *, maximum: int = 20_000) -> str:
        value = arguments.get(name)
        if not isinstance(value, str) or not value.strip() or len(value) > maximum:
            raise ComputerRuntimeError(
                "BROWSER_TOOL_ARGUMENT_INVALID", f"The {name} argument is invalid.", 422
            )
        return value

    async def _observe(self, page: Page) -> dict[str, Any]:
        elements = await page.locator(
            "a,button,input,textarea,select,[role='button'],[role='link'],[role='textbox'],"
            "[contenteditable='true']"
        ).evaluate_all("""
            nodes => nodes.slice(0, 200).map((node, index) => ({
              index,
              tag: node.tagName.toLowerCase(),
              role: node.getAttribute('role'),
              type: node.getAttribute('type'),
              name: node.getAttribute('name'),
              ariaLabel: node.getAttribute('aria-label'),
              text: (node.innerText || node.getAttribute('placeholder') || '').trim().slice(0, 300),
              disabled: Boolean(node.disabled),
              visible: Boolean(node.offsetWidth || node.offsetHeight || node.getClientRects().length)
            }))
        """)
        return {
            "url": page.url[:4096],
            "title": (await page.title())[:500],
            "interactive_elements": elements,
        }

    async def action(
        self, session_id: uuid.UUID, payload: BrowserActionInput, *, request_id: str
    ) -> ActionResult:
        runtime = self._get_runtime(session_id)
        async with runtime.lock:
            self._accept_fence(runtime, payload.fencing_token)
            self._remember_request(runtime, request_id, f"action:{payload.tool}")
            if runtime.status != "ai_controlled":
                raise ComputerRuntimeError(
                    "BROWSER_AI_CONTROL_INACTIVE", "AI control is not active.", 409
                )
            page = await self._page(runtime)
            arguments = payload.arguments
            try:
                result = await self._run_action(
                    runtime, page, payload.tool, arguments, request_id=request_id
                )
            except PlaywrightTimeoutError as error:
                raise ComputerRuntimeError(
                    "BROWSER_ACTION_TIMEOUT", "The browser action timed out.", 504
                ) from error
            except PlaywrightError as error:
                raise ComputerRuntimeError(
                    "BROWSER_ACTION_FAILED", "The browser action failed safely.", 422
                ) from error
            state = await self._state(runtime)
            verification = {
                "status": "observed",
                "url": state.current_url,
                "title": state.page_title,
                "observed_at": state.observed_at,
            }
            self._write_manifest(runtime)
            return ActionResult(
                session=state, tool=payload.tool, result=result, verification=verification
            )

    async def _run_action(
        self, runtime: RuntimeSession, page: Page, tool: str, arguments: dict[str, Any],
        *, request_id: str,
    ) -> dict[str, Any]:
        if tool == "observe":
            return await self._observe(page)
        if tool == "navigate":
            url = self._argument(arguments, "url", maximum=4096)
            await validate_public_url(url)
            await page.goto(url, wait_until="domcontentloaded")
            return {"navigated": True, "url": page.url[:4096]}
        if tool == "click":
            selector = self._argument(arguments, "selector", maximum=1000)
            await page.locator(selector).first.click()
            return {"clicked": True}
        if tool == "type":
            selector = self._argument(arguments, "selector", maximum=1000)
            text = self._argument(arguments, "text")
            await page.locator(selector).first.fill(text)
            return {"typed": True, "character_count": len(text)}
        if tool == "press":
            key = self._argument(arguments, "key", maximum=80)
            if key not in KEY_PATTERN and not (
                key.startswith(("Control+", "Alt+", "Shift+", "Meta+")) and len(key) <= 40
            ):
                raise ComputerRuntimeError(
                    "BROWSER_KEY_BLOCKED", "The keyboard key is not allowed.", 422
                )
            selector = arguments.get("selector")
            if selector is None:
                await page.keyboard.press(key)
            else:
                if not isinstance(selector, str) or not selector.strip() or len(selector) > 1000:
                    raise ComputerRuntimeError(
                        "BROWSER_TOOL_ARGUMENT_INVALID", "The selector argument is invalid.", 422
                    )
                await page.locator(selector).first.press(key)
            return {"pressed": True, "key": key}
        if tool == "wait_for":
            selector = self._argument(arguments, "selector", maximum=1000)
            state = arguments.get("state", "visible")
            if state not in {"attached", "detached", "visible", "hidden"}:
                raise ComputerRuntimeError(
                    "BROWSER_TOOL_ARGUMENT_INVALID", "The wait state is invalid.", 422
                )
            timeout_ms = arguments.get("timeout_ms", 10_000)
            if not isinstance(timeout_ms, int) or not 100 <= timeout_ms <= 30_000:
                raise ComputerRuntimeError(
                    "BROWSER_TOOL_ARGUMENT_INVALID", "The wait timeout is invalid.", 422
                )
            await page.locator(selector).first.wait_for(state=state, timeout=timeout_ms)
            return {"matched": True, "state": state}
        if tool == "extract":
            selector = self._argument(arguments, "selector", maximum=1000)
            attribute = arguments.get("attribute")
            locator = page.locator(selector).first
            if attribute is None:
                value = await locator.inner_text()
            else:
                allowed_attributes = {
                    "href", "title", "alt", "role", "aria-label", "aria-checked",
                    "aria-selected", "aria-expanded",
                }
                if attribute not in allowed_attributes:
                    raise ComputerRuntimeError(
                        "BROWSER_ATTRIBUTE_BLOCKED", "The requested attribute is blocked.", 422
                    )
                value = await locator.get_attribute(attribute)
            return {"value": value[:20_000] if isinstance(value, str) else value}
        if tool == "tab_list":
            state = await self._state(runtime)
            return {"tabs": state.tabs, "active_tab_id": state.active_tab_id}
        if tool == "tab_open":
            if not runtime.context:
                raise ComputerRuntimeError(
                    "BROWSER_SESSION_STOPPED", "The browser session is stopped.", 409
                )
            new_page = await runtime.context.new_page()
            runtime.tab_ids[new_page] = uuid.uuid4().hex
            url = arguments.get("url")
            if url is not None:
                if not isinstance(url, str):
                    raise ComputerRuntimeError(
                        "BROWSER_TOOL_ARGUMENT_INVALID", "The URL argument is invalid.", 422
                    )
                await validate_public_url(url)
                await new_page.goto(url, wait_until="domcontentloaded")
            return {"tab_id": runtime.tab_ids[new_page]}
        if tool == "tab_switch":
            tab_id = self._argument(arguments, "tab_id", maximum=160)
            target = next(
                (candidate for candidate, candidate_id in runtime.tab_ids.items()
                 if candidate_id == tab_id and not candidate.is_closed()),
                None,
            )
            if not target:
                raise ComputerRuntimeError(
                    "BROWSER_TAB_NOT_FOUND", "The browser tab is unavailable.", 404
                )
            await target.bring_to_front()
            return {"active_tab_id": tab_id}
        if tool == "tab_close":
            tab_id = self._argument(arguments, "tab_id", maximum=160)
            target = next(
                (candidate for candidate, candidate_id in runtime.tab_ids.items()
                 if candidate_id == tab_id and not candidate.is_closed()),
                None,
            )
            if not target:
                raise ComputerRuntimeError(
                    "BROWSER_TAB_NOT_FOUND", "The browser tab is unavailable.", 404
                )
            await target.close()
            runtime.tab_ids.pop(target, None)
            return {"closed": True, "tab_id": tab_id}
        if tool == "screenshot":
            image = await page.screenshot(type="jpeg", quality=65, full_page=False)
            artifact_id = uuid.uuid4()
            artifact_directory = (
                self.artifact_root / str(runtime.workspace_id) / str(runtime.session_id)
            ).resolve()
            if not artifact_directory.is_relative_to(self.artifact_root):
                raise ComputerRuntimeError(
                    "BROWSER_ARTIFACT_PATH_INVALID", "The artifact path is invalid.", 422
                )
            artifact_directory.mkdir(parents=True, exist_ok=True)
            filename = f"{artifact_id}.jpg"
            artifact_path = artifact_directory / filename
            temporary = artifact_path.with_suffix(".tmp")
            temporary.write_bytes(image)
            os.replace(temporary, artifact_path)
            return {
                "artifact": {
                    "id": str(artifact_id),
                    "storage_key": (
                        f"{runtime.workspace_id}/{runtime.session_id}/{filename}"
                    ),
                    "sha256": hashlib.sha256(image).hexdigest(),
                    "mime_type": "image/jpeg",
                    "size_bytes": len(image),
                    "metadata": {
                        "width": 1440, "height": 900,
                        "source_request_id": request_id,
                    },
                }
            }
        raise ComputerRuntimeError("BROWSER_TOOL_UNKNOWN", "The browser tool is unknown.", 422)

    async def human_pointer(
        self, session_id: uuid.UUID, payload: HumanPointerInput, *, request_id: str
    ) -> SessionState:
        runtime = self._get_runtime(session_id)
        async with runtime.lock:
            self._accept_fence(runtime, payload.fencing_token)
            self._remember_request(runtime, request_id, f"human:pointer:{payload.action}")
            if runtime.status == "paused":
                runtime.status = "human_controlled"
            if runtime.status != "human_controlled":
                raise ComputerRuntimeError(
                    "BROWSER_HUMAN_CONTROL_INACTIVE", "Human control is not active.", 409
                )
            page = await self._page(runtime)
            if payload.action == "click":
                await page.mouse.click(payload.x, payload.y)
            elif payload.action == "move":
                await page.mouse.move(payload.x, payload.y)
            else:
                await page.mouse.move(payload.x, payload.y)
                await page.mouse.wheel(payload.delta_x, payload.delta_y)
            self._write_manifest(runtime)
            return await self._state(runtime)

    async def human_keyboard(
        self, session_id: uuid.UUID, payload: HumanKeyboardInput, *, request_id: str
    ) -> SessionState:
        runtime = self._get_runtime(session_id)
        async with runtime.lock:
            self._accept_fence(runtime, payload.fencing_token)
            self._remember_request(runtime, request_id, "human:keyboard")
            if runtime.status == "paused":
                runtime.status = "human_controlled"
            if runtime.status != "human_controlled":
                raise ComputerRuntimeError(
                    "BROWSER_HUMAN_CONTROL_INACTIVE", "Human control is not active.", 409
                )
            page = await self._page(runtime)
            if payload.text is not None:
                await page.keyboard.insert_text(payload.text)
            else:
                await page.keyboard.press(payload.key or "")
            self._write_manifest(runtime)
            return await self._state(runtime)

    async def frame(
        self, session_id: uuid.UUID, fencing_token: int, *, request_id: str
    ) -> dict[str, Any]:
        runtime = self._get_runtime(session_id)
        async with runtime.lock:
            self._accept_fence(runtime, fencing_token)
            self._remember_request(runtime, request_id, "frame")
            page = await self._page(runtime)
            image = await page.screenshot(type="jpeg", quality=55, full_page=False)
            return {
                "session_id": str(session_id),
                "fencing_token": runtime.fencing_token,
                "image_base64": base64.b64encode(image).decode("ascii"),
                "mime_type": "image/jpeg",
                "width": 1440,
                "height": 900,
            }

    def read_artifact(
        self, workspace_id: uuid.UUID, session_id: uuid.UUID, artifact_id: uuid.UUID
    ) -> bytes:
        path = (
            self.artifact_root / str(workspace_id) / str(session_id) / f"{artifact_id}.jpg"
        ).resolve()
        if not path.is_relative_to(self.artifact_root):
            raise ComputerRuntimeError(
                "BROWSER_ARTIFACT_PATH_INVALID", "The artifact path is invalid.", 422
            )
        try:
            size = path.stat().st_size
        except OSError as error:
            raise ComputerRuntimeError(
                "BROWSER_ARTIFACT_NOT_FOUND", "The browser artifact is unavailable.", 404
            ) from error
        if not 0 <= size <= 20_000_000:
            raise ComputerRuntimeError(
                "BROWSER_ARTIFACT_INVALID", "The browser artifact is invalid.", 422
            )
        try:
            return path.read_bytes()
        except OSError as error:
            raise ComputerRuntimeError(
                "BROWSER_ARTIFACT_NOT_FOUND", "The browser artifact is unavailable.", 404
            ) from error

    def delete_artifact(
        self, workspace_id: uuid.UUID, session_id: uuid.UUID, artifact_id: uuid.UUID
    ) -> bool:
        path = (
            self.artifact_root / str(workspace_id) / str(session_id) / f"{artifact_id}.jpg"
        ).resolve()
        if not path.is_relative_to(self.artifact_root):
            raise ComputerRuntimeError(
                "BROWSER_ARTIFACT_PATH_INVALID", "The artifact path is invalid.", 422
            )
        try:
            path.unlink()
        except FileNotFoundError:
            return False
        except OSError as error:
            raise ComputerRuntimeError(
                "BROWSER_ARTIFACT_DELETE_FAILED", "The browser artifact could not be deleted.",
                503,
            ) from error
        try:
            path.parent.rmdir()
        except OSError:
            pass
        return True