"""Isolated, privacy-preserving Telegram Web QA setup for DualCoach Tasks 22-25.

This tool deliberately contains no Telegram bot, role-link, message, or product-state
actions. It only presents an official Telegram Web QR page, reports a boolean login
state, and captures redacted browser evidence. It must be run with the installed
Miniconda Playwright runtime on this host.
"""

from __future__ import annotations

import argparse
import asyncio
import contextlib
import io
import json
import os
import re
import stat
import tempfile
import threading
from collections import Counter
from collections.abc import Iterator
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

from PIL import Image
from playwright.async_api import BrowserContext, Page, ViewportSize, async_playwright
from playwright.async_api import Error as PlaywrightError

OFFICIAL_URL = "https://web.telegram.org/"
ROLES = ("customer", "trainer", "owner")
SESSION_PREFIX = "task22-telegram-web-"
SESSION_MARKER = b"task22-telegram-web-qa-v1\n"
DIRECTORY_MODE = 0o700
FILE_MODE = 0o600
VIEWPORT: ViewportSize = {"width": 1280, "height": 900}
DEFAULT_TIMEOUT_SECONDS = 45
DEFAULT_AUTH_TIMEOUT_SECONDS = 900
REDACTION_RGB = (16, 24, 40)


class HarnessError(RuntimeError):
    """A deliberately non-sensitive error suitable for stdout."""

    def __init__(self, code: str, details: dict[str, object] | None = None) -> None:
        super().__init__(code)
        self.code = code
        self.details = details or {}


class LiveDiagnostics:
    """Aggregate only non-sensitive browser lifecycle facts."""

    def __init__(self) -> None:
        self.stage = "startup"
        self._events: Counter[str] = Counter()
        self._hosts: Counter[str] = Counter()
        self._statuses: Counter[str] = Counter()

    @staticmethod
    def _host_bucket(raw_url: str) -> str:
        host = (urlparse(raw_url).hostname or "").lower()
        if host == "web.telegram.org":
            return "official_root"
        if host.endswith(".web.telegram.org"):
            return "official_subdomain"
        return "blocked"

    def record_event(self, event: str, raw_url: str, status: int | None = None) -> None:
        self._events[event] += 1
        self._hosts[self._host_bucket(raw_url)] += 1
        if status is not None and 100 <= status <= 599:
            self._statuses[str(status)] += 1

    def summary(self) -> dict[str, object]:
        return {
            "events": dict(sorted(self._events.items())),
            "hosts": dict(sorted(self._hosts.items())),
            "statuses": dict(sorted(self._statuses.items())),
        }

    def failure_payload(
        self,
        error: BaseException,
        *,
        cleanup_error: BaseException | None = None,
    ) -> dict[str, object]:
        code = {
            "profile_setup": "profile_filesystem_failed",
            "browser_launch": "chromium_launch_failed",
            "guard_install": "browser_guard_failed",
            "navigation": "navigation_failed",
            "state_wait": "login_state_failed",
            "qr_capture": "qr_capture_failed",
            "session_close": "browser_close_failed",
            "cleanup": "profile_cleanup_failed",
        }.get(self.stage, "browser_runtime_failed")
        cause = error.__cause__ if isinstance(error.__cause__, (PlaywrightError, OSError)) else error
        payload: dict[str, object] = {
            "code": code,
            "exception_class": type(cause).__name__,
            "stage": self.stage,
            **self.summary(),
        }
        if cleanup_error is not None:
            payload["cleanup_exception_class"] = type(cleanup_error).__name__
        return payload


@dataclass(frozen=True)
class Capture:
    role: str
    qr_path: Path
    proof_path: Path
    qr_box: dict[str, float]
    qr_quality: bool


@dataclass(frozen=True)
class PageSession:
    context: BrowserContext
    page: Page
    guard: NetworkGuard
    playwright: Any


def emit(value: dict[str, object]) -> None:
    """Print only tool-created metadata, never browser data or exception text."""

    print(json.dumps(value, sort_keys=True, separators=(",", ":")), flush=True)


def secure_mkdir(path: Path) -> None:
    path.mkdir(parents=True, exist_ok=True, mode=DIRECTORY_MODE)
    info = os.lstat(path)
    if not stat.S_ISDIR(info.st_mode) or stat.S_ISLNK(info.st_mode):
        raise HarnessError("unsafe_directory")
    os.chmod(path, DIRECTORY_MODE)


def secure_write(path: Path, payload: bytes) -> None:
    secure_mkdir(path.parent)
    descriptor = os.open(
        path,
        os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0),
        FILE_MODE,
    )
    try:
        written = 0
        while written < len(payload):
            written += os.write(descriptor, payload[written:])
        os.fsync(descriptor)
        os.fchmod(descriptor, FILE_MODE)
    finally:
        os.close(descriptor)


def exact_mode(path: Path, expected: int) -> bool:
    try:
        info = os.lstat(path)
    except FileNotFoundError:
        return False
    return not stat.S_ISLNK(info.st_mode) and stat.S_IMODE(info.st_mode) == expected


def create_session(base_dir: Path | None = None) -> Path:
    base = Path("/tmp") if base_dir is None else base_dir
    if base_dir is not None:
        secure_mkdir(base)
    root = Path(tempfile.mkdtemp(prefix=SESSION_PREFIX, dir=base))
    os.chmod(root, DIRECTORY_MODE)
    secure_write(root / ".task22-session", SESSION_MARKER)
    secure_mkdir(root / "artifacts")
    for role in ROLES:
        secure_mkdir(root / f"{role}-profile")
    return root


def validate_session(root: Path) -> Path:
    root = root.absolute()
    try:
        info = os.lstat(root)
    except FileNotFoundError as exc:
        raise HarnessError("session_missing") from exc
    if (
        root.name.startswith(SESSION_PREFIX) is False
        or not stat.S_ISDIR(info.st_mode)
        or stat.S_ISLNK(info.st_mode)
        or info.st_uid != os.getuid()
        or stat.S_IMODE(info.st_mode) != DIRECTORY_MODE
    ):
        raise HarnessError("unsafe_session")
    marker = root / ".task22-session"
    try:
        marker_value = marker.read_bytes()
    except OSError as exc:
        raise HarnessError("unsafe_session") from exc
    if marker_value != SESSION_MARKER or not exact_mode(marker, FILE_MODE):
        raise HarnessError("unsafe_session")
    return root


def role_profile(root: Path, role: str) -> Path:
    if role not in ROLES:
        raise HarnessError("invalid_role")
    profile = root / f"{role}-profile"
    secure_mkdir(profile)
    if not exact_mode(profile, DIRECTORY_MODE):
        raise HarnessError("unsafe_profile")
    return profile


def secure_remove_tree(path: Path) -> None:
    """Best-effort overwrite then unlink, without traversing symlinks.

    Filesystem snapshots, CoW filesystems, and SSD wear-leveling can prevent physical
    erasure guarantees. This removes the local browser material and overwrites ordinary
    files before unlinking; it never follows a link outside the session root.
    """

    try:
        info = os.lstat(path)
    except FileNotFoundError:
        return
    if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
        raise HarnessError("unsafe_cleanup_target")

    def purge(item: Path) -> None:
        item_info = os.lstat(item)
        if stat.S_ISLNK(item_info.st_mode):
            os.unlink(item)
            return
        if stat.S_ISDIR(item_info.st_mode):
            with os.scandir(item) as children:
                for child in children:
                    purge(Path(child.path))
            os.rmdir(item)
            return
        if stat.S_ISREG(item_info.st_mode):
            descriptor = os.open(
                item,
                os.O_WRONLY | os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0),
            )
            try:
                remaining = os.fstat(descriptor).st_size
                os.lseek(descriptor, 0, os.SEEK_SET)
                block = b"\0" * min(1024 * 1024, max(remaining, 1))
                while remaining:
                    written = os.write(descriptor, block[: min(len(block), remaining)])
                    remaining -= written
                os.fsync(descriptor)
            finally:
                os.close(descriptor)
        os.unlink(item)

    purge(path)


def allowed_url(raw_url: str, fixture_origin: str | None = None) -> bool:
    parsed = urlparse(raw_url)
    if fixture_origin is not None:
        fixture = urlparse(fixture_origin)
        if parsed.scheme == "http" and parsed.scheme == fixture.scheme and parsed.netloc == fixture.netloc:
            return True
    if parsed.scheme not in {"https", "wss"}:
        return False
    host = (parsed.hostname or "").lower()
    return host == "web.telegram.org" or host.endswith(".web.telegram.org")


class NetworkGuard:
    """Block non-Telegram navigation, downloads, dialogs, and clipboard reads."""

    def __init__(
        self,
        fixture_origin: str | None = None,
        diagnostics: LiveDiagnostics | None = None,
    ) -> None:
        self.fixture_origin = fixture_origin
        self.diagnostics = diagnostics
        self.blocked_requests = 0
        self.blocked_downloads = 0
        self.blocked_dialogs = 0
        self.blocked_popups = 0

    def allows(self, raw_url: str) -> bool:
        return allowed_url(raw_url, self.fixture_origin)

    async def install(self, context: BrowserContext) -> None:
        await context.route("**/*", self._route)
        await context.route_web_socket("**/*", self._route_web_socket)
        await context.add_init_script(
            """
            (() => {
              const denied = () => Promise.reject(new DOMException('blocked', 'NotAllowedError'));
              try {
                Object.defineProperty(navigator, 'clipboard', {
                  configurable: false,
                  value: Object.freeze({read: denied, readText: denied}),
                });
              } catch (_) {}
              document.addEventListener('copy', event => event.preventDefault(), true);
              document.addEventListener('cut', event => event.preventDefault(), true);
              document.addEventListener('click', event => {
                const link = event.target instanceof Element && event.target.closest('a[href]');
                if (!link) return;
                const target = new URL(link.href, document.baseURI);
                const isWebTelegram = target.protocol === 'https:' &&
                  (target.hostname === 'web.telegram.org' || target.hostname.endsWith('.web.telegram.org'));
                if (!isWebTelegram) event.preventDefault();
              }, true);
            })();
            """
        )

    async def _route(self, route: Any) -> None:
        if self.allows(route.request.url):
            if self.diagnostics is not None:
                self.diagnostics.record_event("request", route.request.url)
            await route.continue_()
            return
        self.blocked_requests += 1
        if self.diagnostics is not None:
            self.diagnostics.record_event("blocked", route.request.url)
        await route.abort("blockedbyclient")

    async def _route_web_socket(self, route: Any) -> None:
        if self.allows(route.url):
            if self.diagnostics is not None:
                self.diagnostics.record_event("websocket", route.url)
            await route.connect()
            return
        self.blocked_requests += 1
        if self.diagnostics is not None:
            self.diagnostics.record_event("blocked", route.url)
        await route.close()

    def protect_page(self, page: Page) -> None:
        diagnostics = self.diagnostics
        if diagnostics is not None:
            page.on(
                "response",
                lambda response: diagnostics.record_event(
                    "response", response.url, response.status
                ),
            )
            page.on(
                "requestfailed",
                lambda request: diagnostics.record_event("request_failed", request.url),
            )

        def reject_popup(popup: Page) -> None:
            self.blocked_popups += 1
            asyncio.create_task(popup.close())

        def reject_download(download: Any) -> None:
            self.blocked_downloads += 1
            asyncio.create_task(download.cancel())

        def reject_dialog(dialog: Any) -> None:
            self.blocked_dialogs += 1
            asyncio.create_task(dialog.dismiss())

        page.on("popup", reject_popup)
        page.on("download", reject_download)
        page.on("dialog", reject_dialog)


STATE_HELPER = """
(() => {
  const visible = node => {
    if (!(node instanceof Element)) return false;
    const style = getComputedStyle(node);
    if (style.display === 'none' || style.visibility === 'hidden' || Number(style.opacity) === 0) return false;
    const rect = node.getBoundingClientRect();
    return rect.width >= 32 && rect.height >= 32;
  };
  const qrCandidate = () => {
    const nodes = Array.from(document.querySelectorAll("canvas,svg,img,[class*='qr' i],[id*='qr' i]"));
    let best = null;
    let score = -1;
    for (let index = 0; index < nodes.length; index += 1) {
      const node = nodes[index];
      if (!visible(node)) continue;
      const rect = node.getBoundingClientRect();
      const classish = `${node.id || ''} ${node.className || ''} ${node.parentElement?.id || ''} ${node.parentElement?.className || ''}`.toLowerCase();
      const square = 1 - Math.min(1, Math.abs(rect.width - rect.height) / Math.max(rect.width, rect.height));
      const visual = /canvas|svg|img/.test(node.tagName.toLowerCase()) ? 30 : 0;
      const named = /qr|login/.test(classish) ? 100 : 0;
      const sized = rect.width >= 80 && rect.width <= 700 && rect.height >= 80 && rect.height <= 700 ? 30 : 0;
      const current = named + visual + sized + square * 20;
      if (current > score) {
        score = current;
        best = {index, x: rect.x, y: rect.y, width: rect.width, height: rect.height, score: current};
      }
    }
    return best && best.score >= 120 ? best : null;
  };
  window.__task22QrBox = qrCandidate;
  window.__task22DomState = () => {
    if (document.body?.dataset.task22Auth === '1') return 'auth';
    if (qrCandidate()) return 'qr';
    const authenticated = Array.from(document.querySelectorAll("[class*='chat-list' i],[class*='chatlist' i],[class*='sidebar' i],[id*='column-left' i],[id*='dialogs' i]"))
      .some(visible);
    return authenticated ? 'auth' : 'unknown';
  };
})();
"""


async def install_state_helper(page: Page) -> None:
    await page.evaluate(STATE_HELPER)


async def wait_for_state(page: Page, expected: str, timeout_seconds: int) -> str:
    timeout_ms = timeout_seconds * 1000
    try:
        return await page.evaluate(
            """
            ({expected, timeoutMs}) => new Promise((resolve, reject) => {
              const matches = () => {
                const state = window.__task22DomState?.() || 'unknown';
                return expected === 'qr_or_auth' ? state === 'qr' || state === 'auth' : state === expected;
              };
              if (matches()) return resolve(window.__task22DomState());
              const observer = new MutationObserver(() => {
                if (!matches()) return;
                observer.disconnect();
                clearTimeout(timer);
                resolve(window.__task22DomState());
              });
              observer.observe(document.documentElement, {attributes: true, childList: true, subtree: true});
              const timer = setTimeout(() => {
                observer.disconnect();
                reject(new Error('state_timeout'));
              }, timeoutMs);
            })
            """,
            {"expected": expected, "timeoutMs": timeout_ms},
        )
    except PlaywrightError as exc:
        raise HarnessError("state_timeout") from exc


async def qr_box(page: Page) -> dict[str, float]:
    value = await page.evaluate("() => window.__task22QrBox?.() || null")
    if not isinstance(value, dict):
        raise HarnessError("qr_not_found")
    required = ("index", "x", "y", "width", "height")
    if any(key not in value or not isinstance(value[key], (int, float)) for key in required):
        raise HarnessError("qr_not_found")
    if value["width"] < 80 or value["height"] < 80:
        raise HarnessError("qr_not_found")
    return {key: float(value[key]) for key in required}


def qr_visual_quality(png: bytes) -> bool:
    with Image.open(io.BytesIO(png)) as image:
        grayscale = image.convert("L")
        if grayscale.width < 64 or grayscale.height < 64:
            return False
        pixels = [int(pixel) for pixel in grayscale.resize((64, 64)).tobytes()]
    contrast = max(pixels) - min(pixels)
    dark_fraction = sum(pixel < 80 for pixel in pixels) / len(pixels)
    return contrast >= 120 and 0.03 <= dark_fraction <= 0.85


async def capture_qr(page: Page, root: Path, role: str) -> Capture:
    box = await qr_box(page)
    # Crop the in-memory viewport bitmap rather than retaining a page-level QR image or
    # racing a short-lived canvas locator while Telegram refreshes the QR token.
    viewport = await page.screenshot()
    with Image.open(io.BytesIO(viewport)) as image:
        left = max(0, int(box["x"]))
        top = max(0, int(box["y"]))
        right = min(image.width, int(box["x"] + box["width"]))
        bottom = min(image.height, int(box["y"] + box["height"]))
        if right <= left or bottom <= top:
            raise HarnessError("qr_not_found")
        buffer = io.BytesIO()
        image.crop((left, top, right, bottom)).save(buffer, format="PNG")
        crop = buffer.getvalue()
    if not qr_visual_quality(crop):
        raise HarnessError("qr_visual_check_failed")
    qr_path = root / "artifacts" / f"{role}-qr.png"
    secure_write(qr_path, crop)

    proof_path = root / "artifacts" / f"{role}-qr-redacted.png"
    await page.evaluate(
        """
        box => {
          const previous = document.getElementById('__task22_qr_redaction');
          previous?.remove();
          const mask = document.createElement('div');
          mask.id = '__task22_qr_redaction';
          mask.setAttribute('aria-label', 'QR redacted');
          const margin = 10;
          mask.style.cssText = [
            'position:absolute', 'z-index:2147483647', 'pointer-events:none',
            `left:${Math.max(0, box.x + window.scrollX - margin)}px`,
            `top:${Math.max(0, box.y + window.scrollY - margin)}px`,
            `width:${box.width + margin * 2}px`, `height:${box.height + margin * 2}px`,
            'background:#101828', 'border:2px solid #ffffff', 'box-sizing:border-box'
          ].join(';');
          document.body.append(mask);
        }
        """,
        box,
    )
    try:
        proof = await page.screenshot(full_page=True)
    finally:
        await page.evaluate("() => document.getElementById('__task22_qr_redaction')?.remove()")
    secure_write(proof_path, proof)
    return Capture(role, qr_path, proof_path, box, True)


async def capture_authenticated_redaction(page: Page, root: Path, role: str) -> Path:
    proof_path = root / "artifacts" / f"{role}-authenticated-redacted.png"
    await page.evaluate(
        """
        () => {
          const previous = document.getElementById('__task22_full_redaction');
          previous?.remove();
          const mask = document.createElement('div');
          mask.id = '__task22_full_redaction';
          mask.setAttribute('aria-label', 'Authenticated Telegram UI redacted');
          mask.style.cssText = [
            'position:absolute', 'z-index:2147483647', 'pointer-events:none',
            'left:0', 'top:0', `width:${Math.max(document.documentElement.scrollWidth, innerWidth)}px`,
            `height:${Math.max(document.documentElement.scrollHeight, innerHeight)}px`,
            'background:#101828'
          ].join(';');
          document.body.append(mask);
        }
        """
    )
    try:
        proof = await page.screenshot(full_page=True)
    finally:
        await page.evaluate("() => document.getElementById('__task22_full_redaction')?.remove()")
    secure_write(proof_path, proof)
    return proof_path


async def open_page_session(
    profile: Path,
    *,
    fixture_origin: str | None,
    headless: bool,
    diagnostics: LiveDiagnostics | None = None,
) -> PageSession:
    playwright = await async_playwright().start()
    if diagnostics is not None:
        diagnostics.stage = "browser_launch"
    try:
        context = await playwright.chromium.launch_persistent_context(
            str(profile),
            headless=headless,
            viewport=VIEWPORT,
            accept_downloads=False,
            permissions=[],
        )
    except BaseException:
        await playwright.stop()
        raise
    if diagnostics is not None:
        diagnostics.stage = "guard_install"
    guard = NetworkGuard(fixture_origin, diagnostics)
    await guard.install(context)
    page = context.pages[0] if context.pages else await context.new_page()
    guard.protect_page(page)
    return PageSession(context, page, guard, playwright)


async def close_page_session(session: PageSession) -> None:
    with contextlib.suppress(PlaywrightError):
        await session.context.close()
    await session.playwright.stop()


async def load(
    page: Page,
    url: str,
    timeout_seconds: int,
    diagnostics: LiveDiagnostics | None = None,
) -> None:
    if not allowed_url(url) and not url.startswith("http://127.0.0.1:"):
        raise HarnessError("navigation_rejected")
    if diagnostics is not None:
        diagnostics.stage = "navigation"
    try:
        await page.goto(url, wait_until="domcontentloaded", timeout=timeout_seconds * 1000)
        await install_state_helper(page)
    except PlaywrightError as exc:
        raise HarnessError("navigation_failed") from exc


async def prepare_role(
    root: Path,
    role: str,
    *,
    url: str,
    fixture_origin: str | None,
    timeout_seconds: int,
    headless: bool,
    diagnostics: LiveDiagnostics | None = None,
) -> tuple[PageSession, str, Capture | None]:
    profile = role_profile(root, role)
    session = await open_page_session(
        profile,
        fixture_origin=fixture_origin,
        headless=headless,
        diagnostics=diagnostics,
    )
    try:
        await load(session.page, url, timeout_seconds, diagnostics)
        if diagnostics is not None:
            diagnostics.stage = "state_wait"
        state = await wait_for_state(session.page, "qr_or_auth", timeout_seconds)
        if state == "auth":
            return session, state, None
        if state != "qr":
            raise HarnessError("login_ui_not_detected")
        if diagnostics is not None:
            diagnostics.stage = "qr_capture"
        return session, state, await capture_qr(session.page, root, role)
    except (HarnessError, PlaywrightError, OSError):
        with contextlib.suppress(PlaywrightError, OSError):
            await close_page_session(session)
        raise


async def command_prepare(args: argparse.Namespace) -> None:
    root = validate_session(Path(args.session_root))
    if args.fixture_url is not None:
        raise HarnessError("fixture_only_available_to_self_test")
    session, state, capture = await prepare_role(
        root,
        args.role,
        url=OFFICIAL_URL,
        fixture_origin=None,
        timeout_seconds=args.timeout_seconds,
        headless=not args.headed,
    )
    try:
        if state == "auth":
            emit({"authenticated": True, "role": args.role, "state": "paused" if args.interactive else "closed"})
            if args.interactive:
                await asyncio.Event().wait()
            return
        if capture is None:
            raise HarnessError("qr_not_found")
        emit(
            {
                "authenticated": False,
                "proof": str(capture.proof_path),
                "qr": str(capture.qr_path),
                "role": args.role,
                "state": "qr_ready",
            }
        )
        if not args.interactive:
            return
        await wait_for_state(session.page, "auth", args.auth_timeout_seconds)
        emit({"authenticated": True, "role": args.role, "state": "paused"})
        await asyncio.Event().wait()
    finally:
        await close_page_session(session)


async def command_capture(args: argparse.Namespace) -> None:
    root = validate_session(Path(args.session_root))
    profile = role_profile(root, args.role)
    session = await open_page_session(profile, fixture_origin=None, headless=not args.headed)
    try:
        await load(session.page, OFFICIAL_URL, args.timeout_seconds)
        state = await wait_for_state(session.page, "qr_or_auth", args.timeout_seconds)
        if state != "auth":
            emit({"authenticated": False, "role": args.role, "state": "closed"})
            return
        proof = await capture_authenticated_redaction(session.page, root, args.role)
        emit({"authenticated": True, "proof": str(proof), "role": args.role, "state": "closed"})
    finally:
        await close_page_session(session)


async def clear_local_site_data(page: Page, context: BrowserContext) -> bool:
    """Clear only browser-local Telegram state; no account identity is read or emitted."""

    try:
        await page.evaluate(
            """
            async () => {
              localStorage.clear();
              sessionStorage.clear();
              await Promise.all((await caches.keys()).map(key => caches.delete(key)));
              if (indexedDB.databases) {
                const databases = await indexedDB.databases();
                await Promise.all(databases.filter(item => item.name).map(item => new Promise(resolve => {
                  const request = indexedDB.deleteDatabase(item.name);
                  request.onsuccess = request.onerror = request.onblocked = () => resolve();
                })));
              }
            }
            """
        )
        await context.clear_cookies()
    except PlaywrightError:
        return False
    return True


async def attempt_web_logout(page: Page, timeout_seconds: int) -> bool:
    """Attempt the visible local logout control only when cleanup explicitly requests it."""

    try:
        await page.goto(f"{OFFICIAL_URL}k/#settings", wait_until="domcontentloaded", timeout=timeout_seconds * 1000)
        await install_state_helper(page)
        logout = page.get_by_text(re.compile(r"^(log\s*out|로그아웃)$", re.IGNORECASE))
        if await logout.count() != 1 or not await logout.is_visible():
            return False
        await logout.click(timeout=timeout_seconds * 1000)
    except PlaywrightError:
        return False
    return True


async def command_cleanup(args: argparse.Namespace) -> None:
    root = validate_session(Path(args.session_root))
    local_clear = False
    remote_logout_attempted = False
    if args.logout:
        for role in ROLES:
            profile = role_profile(root, role)
            session = await open_page_session(profile, fixture_origin=None, headless=True)
            try:
                await load(session.page, OFFICIAL_URL, args.timeout_seconds)
                state = await wait_for_state(session.page, "qr_or_auth", args.timeout_seconds)
                if state == "auth":
                    remote_logout_attempted = await attempt_web_logout(session.page, args.timeout_seconds) or remote_logout_attempted
                    local_clear = await clear_local_site_data(session.page, session.context) or local_clear
            finally:
                await close_page_session(session)
    secure_remove_tree(root)
    emit(
        {
            "local_browser_state_cleared": local_clear,
            "profile_directories_removed": True,
            "remote_logout_attempted": remote_logout_attempted,
            "state": "cleaned",
        }
    )


class FixtureHandler(BaseHTTPRequestHandler):
    secret = "fixture-secret-must-never-reach-output"

    def do_GET(self) -> None:
        payload = f"""<!doctype html>
<html><head><meta charset='utf-8'><style>
body {{ margin:0; background:#f5f7fb; }}
#shell {{ width:920px; height:680px; margin:40px auto; background:white; border:1px solid #ccd5e1; }}
#fixture-qr {{ width:280px; height:280px; margin:130px auto; display:block; }}
#fixture-main {{ display:none; width:100%; height:100%; }}
</style></head><body data-fixture-secret='{self.secret}'>
<main id='shell'><canvas id='fixture-qr' class='auth-qr' width='280' height='280'></canvas>
<section id='fixture-main' class='chat-list'></section><button id='fixture-auth'>Fixture authentication event</button></main>
<script>
const canvas = document.getElementById('fixture-qr'); const context = canvas.getContext('2d');
context.fillStyle = '#fff'; context.fillRect(0, 0, 280, 280); context.fillStyle = '#000';
for (let y = 16; y < 264; y += 8) for (let x = 16; x < 264; x += 8) if (((x / 8) * 7 + (y / 8) * 11) % 5 < 2) context.fillRect(x, y, 8, 8);
document.getElementById('fixture-auth').addEventListener('click', () => {{
  document.body.dataset.task22Auth = '1'; canvas.style.display = 'none'; document.getElementById('fixture-main').style.display = 'block';
}});
</script></body></html>""".encode()
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, format: str, *args: object) -> None:
        return


@contextlib.contextmanager
def fixture_server() -> Iterator[str]:
    server = ThreadingHTTPServer(("127.0.0.1", 0), FixtureHandler)
    worker = threading.Thread(target=server.serve_forever, daemon=True)
    worker.start()
    try:
        yield f"http://127.0.0.1:{server.server_address[1]}/"
    finally:
        server.shutdown()
        worker.join(timeout=5)
        server.server_close()


def assert_redaction(capture: Capture) -> None:
    with Image.open(capture.proof_path) as image:
        x = max(0, int(capture.qr_box["x"] + capture.qr_box["width"] / 2))
        y = max(0, int(capture.qr_box["y"] + capture.qr_box["height"] / 2))
        pixel = image.convert("RGB").getpixel((x, y))
    if pixel != REDACTION_RGB:
        raise HarnessError("fixture_redaction_failed")


async def self_test() -> None:
    with tempfile.TemporaryDirectory(prefix="task22-harness-test-") as temporary, fixture_server() as fixture_url:
        root = create_session(Path(temporary))
        fixture_origin = fixture_url.removesuffix("/")
        customer_session, customer_state, customer_capture = await prepare_role(
            root,
            "customer",
            url=fixture_url,
            fixture_origin=fixture_origin,
            timeout_seconds=10,
            headless=True,
        )
        await close_page_session(customer_session)
        trainer_session, trainer_state, trainer_capture = await prepare_role(
            root,
            "trainer",
            url=fixture_url,
            fixture_origin=fixture_origin,
            timeout_seconds=10,
            headless=True,
        )
        await close_page_session(trainer_session)
        if customer_state != "qr" or trainer_state != "qr" or customer_capture is None or trainer_capture is None:
            raise HarnessError("fixture_qr_state_failed")
        if not all(exact_mode(path, FILE_MODE) for path in (customer_capture.qr_path, customer_capture.proof_path, trainer_capture.qr_path, trainer_capture.proof_path)):
            raise HarnessError("fixture_permissions_failed")
        if not all(exact_mode(root / f"{role}-profile", DIRECTORY_MODE) for role in ROLES):
            raise HarnessError("fixture_profile_permissions_failed")
        assert_redaction(customer_capture)
        with Image.open(customer_capture.qr_path) as image:
            if image.width >= 920 or image.height >= 680:
                raise HarnessError("fixture_qr_crop_failed")

        customer_profile = role_profile(root, "customer")
        customer = await open_page_session(customer_profile, fixture_origin=fixture_origin, headless=True)
        try:
            await load(customer.page, fixture_url, 10)
            await customer.page.evaluate("() => localStorage.setItem('fixture-isolation', 'customer')")
        finally:
            await close_page_session(customer)
        trainer_profile = role_profile(root, "trainer")
        trainer = await open_page_session(trainer_profile, fixture_origin=fixture_origin, headless=True)
        try:
            await load(trainer.page, fixture_url, 10)
            if await trainer.page.evaluate("() => localStorage.getItem('fixture-isolation')") is not None:
                raise HarnessError("fixture_role_isolation_failed")
            authenticated_wait = asyncio.create_task(wait_for_state(trainer.page, "auth", 10))
            await trainer.page.locator("#fixture-auth").click()
            if await authenticated_wait != "auth":
                raise HarnessError("fixture_auth_event_failed")
            try:
                await trainer.page.goto("https://example.invalid/blocked", timeout=5_000)
            except PlaywrightError:
                pass
            if trainer.guard.blocked_requests < 1:
                raise HarnessError("fixture_navigation_guard_failed")
        finally:
            await close_page_session(trainer)

        rendered = io.StringIO()
        with contextlib.redirect_stdout(rendered):
            emit({"authenticated": False, "role": "customer", "state": "qr_ready"})
            await command_cleanup(
                argparse.Namespace(session_root=root, logout=False, timeout_seconds=10)
            )
        if FixtureHandler.secret in rendered.getvalue():
            raise HarnessError("fixture_secret_output_failed")
        if root.exists():
            raise HarnessError("fixture_cleanup_failed")


async def command_self_test(_args: argparse.Namespace) -> None:
    await self_test()
    emit(
        {
            "cleanup": "pass",
            "fixture": "pass",
            "permissions": "pass",
            "role_isolation": "pass",
            "state": "pass",
        }
    )


async def command_live_smoke(args: argparse.Namespace) -> None:
    proof_path = Path(args.proof_path).absolute()
    qr_path = Path(args.qr_path).absolute()
    diagnostics = LiveDiagnostics()
    root: Path | None = None
    primary: tuple[BaseException, str] | None = None
    cleanup_error: BaseException | None = None
    capture: Capture | None = None
    try:
        diagnostics.stage = "profile_setup"
        root = create_session()
        session, state, capture = await prepare_role(
            root,
            "customer",
            url=OFFICIAL_URL,
            fixture_origin=None,
            timeout_seconds=args.timeout_seconds,
            headless=True,
            diagnostics=diagnostics,
        )
        diagnostics.stage = "session_close"
        await close_page_session(session)
        if state != "qr" or capture is None or not capture.qr_quality:
            raise HarnessError("live_qr_ui_not_detected")
        assert_redaction(capture)
        secure_write(proof_path, capture.proof_path.read_bytes())
        secure_write(qr_path, capture.qr_path.read_bytes())
    except (
        HarnessError,
        PlaywrightError,
        OSError,
        AttributeError,
        KeyError,
        RuntimeError,
        TypeError,
        ValueError,
    ) as error:
        primary = (error, diagnostics.stage)
    finally:
        diagnostics.stage = "cleanup"
        if root is not None:
            try:
                secure_remove_tree(root)
            except (HarnessError, OSError) as error:
                cleanup_error = error

    if primary is not None or cleanup_error is not None:
        for path in (proof_path, qr_path):
            with contextlib.suppress(FileNotFoundError):
                path.unlink()
        error, stage = primary if primary is not None else (cleanup_error, "cleanup")
        assert error is not None
        diagnostics.stage = stage
        payload = diagnostics.failure_payload(error, cleanup_error=cleanup_error)
        code = str(payload.pop("code"))
        raise HarnessError(code, payload) from error

    if capture is None:
        raise HarnessError("live_qr_ui_not_detected")
    emit(
        {
            **diagnostics.summary(),
            "profile_removed": True,
            "proof": str(proof_path),
            "qr": str(qr_path),
            "state": "qr_ready",
        }
    )


def parser() -> argparse.ArgumentParser:
    result = argparse.ArgumentParser(description=__doc__)
    commands = result.add_subparsers(dest="command", required=True)

    init = commands.add_parser("init", help="create one owner-only Task 22 browser session")
    init.add_argument("--base-dir", type=Path)

    prepare = commands.add_parser("prepare", help="present one isolated official QR login")
    prepare.add_argument("role", choices=ROLES)
    prepare.add_argument("--session-root", required=True, type=Path)
    prepare.add_argument("--interactive", action="store_true")
    prepare.add_argument("--headed", action="store_true", help="requires --interactive and a local display")
    prepare.add_argument("--timeout-seconds", type=int, default=DEFAULT_TIMEOUT_SECONDS)
    prepare.add_argument("--auth-timeout-seconds", type=int, default=DEFAULT_AUTH_TIMEOUT_SECONDS)
    prepare.add_argument("--fixture-url", help=argparse.SUPPRESS)

    capture = commands.add_parser("capture", help="verify boolean auth and save fully redacted UI proof")
    capture.add_argument("role", choices=ROLES)
    capture.add_argument("--session-root", required=True, type=Path)
    capture.add_argument("--timeout-seconds", type=int, default=DEFAULT_TIMEOUT_SECONDS)
    capture.add_argument("--headed", action="store_true")

    cleanup = commands.add_parser("cleanup", help="remove all Task 22 browser profiles and QR artifacts")
    cleanup.add_argument("--session-root", required=True, type=Path)
    cleanup.add_argument("--logout", action="store_true", help="after Task 25 only: attempt local Web logout before removal")
    cleanup.add_argument("--timeout-seconds", type=int, default=DEFAULT_TIMEOUT_SECONDS)

    commands.add_parser("self-test", help="run only against a synthetic loopback fixture")
    live = commands.add_parser("live-smoke", help="one ephemeral unauthenticated official QR load")
    live.add_argument("--proof-path", required=True)
    live.add_argument("--qr-path", required=True)
    live.add_argument("--timeout-seconds", type=int, default=DEFAULT_TIMEOUT_SECONDS)
    return result


async def dispatch(args: argparse.Namespace) -> None:
    if args.command == "init":
        root = create_session(args.base_dir)
        emit({"session_root": str(root), "state": "ready"})
        return
    if getattr(args, "timeout_seconds", 1) <= 0 or getattr(args, "auth_timeout_seconds", 1) <= 0:
        raise HarnessError("invalid_timeout")
    if args.command == "prepare":
        if args.headed and not args.interactive:
            raise HarnessError("headed_requires_interactive")
        await command_prepare(args)
        return
    if args.command == "capture":
        await command_capture(args)
        return
    if args.command == "cleanup":
        await command_cleanup(args)
        return
    if args.command == "self-test":
        await command_self_test(args)
        return
    if args.command == "live-smoke":
        await command_live_smoke(args)
        return
    raise HarnessError("invalid_command")


def main() -> int:
    args = parser().parse_args()
    try:
        asyncio.run(dispatch(args))
    except HarnessError as exc:
        emit({"code": exc.code, **exc.details, "state": "error"})
        return 2
    except KeyboardInterrupt:
        return 130
    except (PlaywrightError, OSError):
        emit({"code": "browser_or_filesystem_failure", "state": "error"})
        return 2
    except (AttributeError, KeyError, RuntimeError, TypeError, ValueError):
        emit({"code": "internal_failure", "state": "error"})
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())