jarca-ledger-mcp/scripts/streamlit_browser_evidence.py

#!/usr/bin/env python3
"""Capture local Streamlit browser evidence with Chrome and a synthetic DB.

This script is intentionally local-only. It creates a synthetic fixture database,
starts Streamlit against that database, and uses an installed Chrome binary to
save page screenshots and DOM snapshots. It must not target the operating ledger
DB or real financial statement files.
"""

from __future__ import annotations

import argparse
import asyncio
import base64
import json
import os
import platform
import shutil
import signal
import subprocess
import sys
import time
import hashlib
import math
import secrets
import ssl
import urllib.parse
import urllib.request
from datetime import datetime
from pathlib import Path
from typing import Any

import websockets

from jarca_ledger_mcp import service
from jarca_ledger_mcp.db import init_db

DEFAULT_PAGES = ["요약", "분류 규칙", "확인필요 큐", "Import"]
DEFAULT_CHROME = "/usr/bin/google-chrome-stable"
DEFAULT_STREAMLIT = Path(".venv/bin/streamlit")
DEFAULT_OUTPUT_ROOT = Path("docs/batches/bat-014_20260703_adapter-surface-narrowing-baseline-refresh/evidence")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Capture Streamlit browser screenshots and DOM snapshots with a synthetic DB.")
    parser.add_argument("--chrome", default=os.getenv("CHROME_BIN", DEFAULT_CHROME), help="Chrome executable path")
    parser.add_argument("--streamlit", default=str(DEFAULT_STREAMLIT), help="Streamlit executable path")
    parser.add_argument("--port", type=int, default=18501, help="Local Streamlit port")
    parser.add_argument("--host", default="127.0.0.1", help="Local Streamlit host")
    parser.add_argument("--output-dir", default="", help="Evidence output directory. Defaults under docs/batches/BAT-004/evidence/run-<timestamp>.")
    parser.add_argument("--page", action="append", dest="pages", help="Page to capture. Repeatable. Defaults to core E2E evidence pages.")
    parser.add_argument("--keep-server", action="store_true", help="Leave Streamlit running after capture for manual inspection.")
    parser.add_argument("--mode", choices=("capture", "transition"), default="capture", help="Capture mode")
    parser.add_argument("--cert-file", default="", help="TLS certificate for transition mode")
    parser.add_argument("--key-file", default="", help="TLS private key for transition mode")
    parser.add_argument("--from-page", default="요약")
    parser.add_argument("--to-page", default="분류 규칙")
    parser.add_argument("--interval-ms", type=int, default=100)
    parser.add_argument("--frames", type=int, default=8)
    return parser.parse_args()


def slugify_page(name: str) -> str:
    mapping = {
        "분류 규칙": "rules",
        "확인필요 큐": "transfer-review",
        "Import": "import",
        "계좌": "accounts",
        "요약": "overview",
        "거래내역": "transactions",
        "미분류": "uncategorized",
        "카테고리": "categories",
    }
    return mapping.get(name, urllib.parse.quote(name, safe="").replace("%", "").lower() or "page")


def require_executable(path: str, label: str) -> str:
    resolved = shutil.which(path) if os.path.sep not in path else path
    if not resolved or not Path(resolved).exists():
        raise SystemExit(f"{label} executable not found: {path}")
    return resolved


def assert_synthetic_db_path(path: Path) -> None:
    resolved = path.expanduser().resolve()
    operating = Path.home() / ".local" / "share" / "jarca" / "ledger.sqlite3"
    if resolved == operating.resolve():
        raise SystemExit("Refusing to use the operating ledger DB for browser evidence")
    if not str(resolved).endswith("fixture.sqlite3"):
        raise SystemExit(f"Synthetic evidence DB must end with fixture.sqlite3: {resolved}")


def seed_fixture_db(db_path: Path) -> dict[str, Any]:
    assert_synthetic_db_path(db_path)
    if db_path.exists():
        db_path.unlink()
    init_db(str(db_path))
    service.seed_default_ledger_categories(db_path=str(db_path))
    service.seed_default_category_rules(db_path=str(db_path))

    account = service.create_ledger_account(
        "Synthetic Evidence Checking",
        institution="Synthetic Bank",
        account_number_masked="SYNTH-0000",
        holder_name="Synthetic Holder",
        last_balance=1_250_000,
        last_synced_at="2026-06-30T23:59:00+09:00",
        statement_status="partial_import",
        liquidity_bucket="cash",
        memo="browser evidence synthetic fixture",
        db_path=str(db_path),
    )["account"]
    account_id = account["id"]
    service.create_ledger_account(
        "Synthetic Investment Account",
        type="investment",
        institution="Synthetic Securities",
        last_balance=5_000_000,
        last_synced_at="2026-06-30T23:59:00+09:00",
        liquidity_bucket="investment",
        memo="browser evidence investment asset",
        db_path=str(db_path),
    )
    service.create_ledger_account(
        "Synthetic Mortgage Balance",
        type="loan",
        institution="Synthetic Bank",
        last_balance=2_000_000,
        last_synced_at="2026-06-30T23:59:00+09:00",
        liquidity_bucket="loan",
        memo="browser evidence liability",
        db_path=str(db_path),
    )
    service.create_ledger_account(
        "Synthetic Appraisal Pending",
        type="other",
        institution="Manual",
        liquidity_bucket="real_estate",
        memo="browser evidence missing last_balance warning",
        db_path=str(db_path),
    )
    service.create_ledger_budget("2026-06", "식비", 20_000, memo="browser evidence budget", db_path=str(db_path))

    tx_specs = [
        dict(transaction_date="2026-06-25", type="income", amount=2_000_000, merchant="SYNTHETIC_SALARY", category="수입", subcategory="급여", source="manual", memo="browser evidence income"),
        dict(transaction_date="2026-06-26", type="expense", amount=35_000, merchant="SYNTHETIC_DINING", category="식비", subcategory="외식", source="manual", memo="browser evidence expense"),
        dict(transaction_date="2026-06-27", type="expense", amount=18_500, merchant="SYNTHETIC_MART", category="", subcategory="", source="manual", memo="browser evidence uncategorized", raw_summary="synthetic grocery"),
        dict(transaction_date="2026-06-28", type="expense", amount=4_800, merchant="SYNTHETIC_CAFE", category="", subcategory="", source="manual", memo="browser evidence uncategorized", raw_summary="synthetic cafe"),
        dict(transaction_date="2026-06-29", type="transfer", amount=120_000, merchant="SYNTHETIC_TRANSFER", category="이체", subcategory="확인필요", source="kb_bank_pdf", counterparty="SYNTHETIC_TOSS", memo="cross_transfer_review: synthetic browser evidence", raw_summary="synthetic transfer", balance_after=1_250_000),
    ]
    transactions = []
    for index, spec in enumerate(tx_specs, start=1):
        result = service.create_ledger_transaction(account_id=account_id, external_id=f"browser-evidence-{index}", db_path=str(db_path), **spec)
        if not result.get("success"):
            raise SystemExit(f"Fixture transaction seed failed: {result}")
        transactions.append(result["transaction"]["id"])

    return {"account_id": account_id, "transaction_ids": transactions, "db_path": str(db_path)}


def wait_for_http(url: str, timeout_seconds: int = 45, ssl_context: ssl.SSLContext | None = None) -> None:
    """Wait for HTTP readiness, using verified TLS unless explicitly overridden."""
    deadline = time.time() + timeout_seconds
    last_error: str | None = None
    while time.time() < deadline:
        try:
            with urllib.request.urlopen(url, timeout=2, context=ssl_context) as response:
                if response.status == 200:
                    return
                last_error = f"HTTP {response.status}"
        except Exception as exc:  # noqa: BLE001 - diagnostics returned in manifest
            last_error = str(exc)
        time.sleep(1)
    raise SystemExit(f"Streamlit did not become ready at {url}: {last_error}")


def read_json_url(url: str, timeout: int = 2) -> Any:
    with urllib.request.urlopen(url, timeout=timeout) as response:
        return json.loads(response.read().decode("utf-8"))


def wait_for_devtools(port: int, timeout_seconds: int = 15) -> str:
    deadline = time.time() + timeout_seconds
    last_error: str | None = None
    while time.time() < deadline:
        try:
            targets = read_json_url(f"http://127.0.0.1:{port}/json/list")
            for target in targets:
                if target.get("type") == "page" and target.get("webSocketDebuggerUrl"):
                    return target["webSocketDebuggerUrl"]
            last_error = "no page target"
        except Exception as exc:  # noqa: BLE001 - Chrome startup diagnostics
            last_error = str(exc)
        time.sleep(0.25)
    raise SystemExit(f"Chrome DevTools endpoint did not become ready on port {port}: {last_error}")


def remove_tree_with_retries(path: Path, attempts: int = 5, delay_seconds: float = 0.5) -> None:
    for attempt in range(1, attempts + 1):
        shutil.rmtree(path, ignore_errors=True)
        if not path.exists():
            return
        time.sleep(delay_seconds * attempt)
    if path.exists():
        raise SystemExit(f"temporary Chrome profile cleanup failed: {path}")


class CdpClient:
    def __init__(self, ws: websockets.ClientConnection) -> None:
        self.ws = ws
        self.next_id = 1
        self.events: list[dict[str, Any]] = []

    async def call(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
        command_id = self.next_id
        self.next_id += 1
        await self.ws.send(json.dumps({"id": command_id, "method": method, "params": params or {}}))
        while True:
            raw = await self.ws.recv()
            payload = json.loads(raw)
            if payload.get("id") == command_id:
                if "error" in payload:
                    raise RuntimeError(f"CDP {method} failed: {payload['error']}")
                return payload.get("result", {})
            if payload.get("method"):
                self.events.append(payload)


async def cdp_capture(ws_url: str, url: str, expected_text: str, screenshot: Path, snapshot: Path) -> dict[str, Any]:
    async with websockets.connect(ws_url, max_size=16 * 1024 * 1024) as ws:
        cdp = CdpClient(ws)
        await cdp.call("Page.enable")
        await cdp.call("Runtime.enable")
        await cdp.call(
            "Emulation.setDeviceMetricsOverride",
            {"width": 1440, "height": 1400, "deviceScaleFactor": 1, "mobile": False},
        )
        await cdp.call("Page.navigate", {"url": url})

        inner_text = ""
        deadline = time.time() + 45
        while time.time() < deadline:
            result = await cdp.call(
                "Runtime.evaluate",
                {"expression": "document.body ? document.body.innerText : ''", "returnByValue": True},
            )
            inner_text = result.get("result", {}).get("value", "") or ""
            if expected_text in inner_text:
                break
            await asyncio.sleep(1)
        if expected_text not in inner_text:
            raise SystemExit(f"Browser page did not render expected text {expected_text!r}; last text starts with {inner_text[:200]!r}")

        html_result = await cdp.call(
            "Runtime.evaluate",
            {"expression": "document.documentElement.outerHTML", "returnByValue": True},
        )
        snapshot.write_text(html_result.get("result", {}).get("value", ""), encoding="utf-8")
        image_result = await cdp.call("Page.captureScreenshot", {"format": "png", "fromSurface": True, "captureBeyondViewport": True})
        screenshot.write_bytes(base64.b64decode(image_result["data"]))
        return {"inner_text_head": inner_text[:500]}


def capture_page(chrome: str, base_url: str, page: str, output_dir: Path, devtools_port: int) -> dict[str, Any]:
    slug = slugify_page(page)
    url = f"{base_url}/?page={urllib.parse.quote(page)}"
    screenshot = output_dir / f"{slug}.png"
    snapshot = output_dir / f"{slug}.dom.html"

    profile_dir = output_dir / f"chrome-profile-{slug}"
    chrome_cmd = [
        chrome,
        "--headless=new",
        "--no-sandbox",
        "--disable-dev-shm-usage",
        "--disable-gpu",
        "--disable-software-rasterizer",
        "--hide-scrollbars",
        f"--remote-debugging-port={devtools_port}",
        f"--user-data-dir={profile_dir}",
        "--window-size=1440,1400",
        "about:blank",
    ]
    chrome_proc = subprocess.Popen(chrome_cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    try:
        ws_url = wait_for_devtools(devtools_port)
        cdp_info = asyncio.run(cdp_capture(ws_url, url, page, screenshot, snapshot))
    finally:
        if chrome_proc.poll() is None:
            chrome_proc.terminate()
            try:
                chrome_proc.wait(timeout=8)
            except subprocess.TimeoutExpired:
                chrome_proc.kill()
                chrome_proc.wait(timeout=5)
        remove_tree_with_retries(profile_dir)

    if not screenshot.exists() or screenshot.stat().st_size <= 0:
        raise SystemExit(f"Screenshot was not created for {page}: {screenshot}")
    if page not in snapshot.read_text(encoding="utf-8", errors="ignore"):
        raise SystemExit(f"DOM snapshot does not contain expected page title {page!r}: {snapshot}")

    return {
        "page": page,
        "url": url,
        "screenshot": str(screenshot),
        "screenshot_bytes": screenshot.stat().st_size,
        "snapshot": str(snapshot),
        "snapshot_bytes": snapshot.stat().st_size,
        "devtools_port": devtools_port,
        "inner_text_head": cdp_info.get("inner_text_head", ""),
    }


def redact_secrets(value: Any, secrets_to_redact: tuple[str, ...] = ()) -> Any:
    """Return JSON-safe evidence with credential-like values removed."""
    if isinstance(value, dict):
        return {key: redact_secrets(item, secrets_to_redact) for key, item in value.items()}
    if isinstance(value, list):
        return [redact_secrets(item, secrets_to_redact) for item in value]
    if isinstance(value, str):
        result = value
        for secret in secrets_to_redact:
            if secret:
                result = result.replace(secret, "[REDACTED]")
        return result
    return value


def _validate_surface_records(records: Any, label: str) -> None:
    required = ("stAppViewContainer", "stSidebar")
    if not isinstance(records, dict) or any(not isinstance(records.get(name), dict) for name in required):
        raise ValueError(f"transition observations require surface records for {label}")
    for name in required:
        surface = records[name]
        if any(key not in surface or surface[key] is None for key in ("backgroundImage", "backgroundColor", "visibility", "display", "opacity", "rendered")):
            raise ValueError(f"transition observations require complete surface records for {label}")
        if not str(surface["backgroundImage"]).strip() or str(surface["backgroundImage"]).strip().lower() == "none":
            raise ValueError(f"transition observations reject empty surface background for {label}")
        if not str(surface["backgroundColor"]).strip():
            raise ValueError(f"transition observations reject empty surface background color for {label}")
        try:
            opacity = float(surface["opacity"])
        except (TypeError, ValueError) as exc:
            raise ValueError(f"transition observations require valid surface opacity for {label}") from exc
        if not math.isfinite(opacity) or surface["visibility"] != "visible" or surface["display"] == "none" or opacity <= 0 or surface["rendered"] is not True:
            raise ValueError(f"transition observations require visible rendered surfaces for {label}")


def validate_transition_manifest(manifest: dict[str, Any], secrets_to_redact: tuple[str, ...] = ()) -> None:
    """Fail closed unless transition evidence contains every required observation."""
    serialized = json.dumps(manifest, ensure_ascii=False)
    if any(secret and secret in serialized for secret in secrets_to_redact):
        raise ValueError("secret value present in transition manifest")
    if manifest.get("success") is not True:
        raise ValueError("transition manifest is not successful")
    observations = manifest.get("observations") or {}
    frames = observations.get("frames")
    if not isinstance(frames, list) or len(frames) < 2:
        raise ValueError("transition observations require at least two frames")
    for index, frame in enumerate(frames):
        _validate_surface_records(frame.get("surfaces"), f"frame {index}")
    document_events = observations.get("document_navigation_events")
    if not isinstance(document_events, list) or document_events:
        raise ValueError("transition observations require zero document navigation events")
    shell = observations.get("stable_shell") or {}
    stable_fields = ("document_marker", "sidebar_marker", "background")
    if any(not isinstance(shell.get(key), dict) for key in stable_fields):
        raise ValueError("transition observations require pre/post stable shell markers")
    if any(
        not isinstance(values.get(position), str) or not values[position].strip()
        for values in (shell[key] for key in stable_fields)
        for position in ("pre", "post")
    ):
        raise ValueError("transition observations require non-empty stable shell markers")
    if any(shell[key]["pre"] != shell[key]["post"] for key in stable_fields):
        raise ValueError("transition observations require stable shell identity and background")
    stable_surfaces = shell.get("surfaces")
    if not isinstance(stable_surfaces, dict) or stable_surfaces.get("pre") != frames[0].get("surfaces"):
        raise ValueError("transition observations require stable shell surface records")
    _validate_surface_records(stable_surfaces.get("pre"), "stable shell pre")
    _validate_surface_records(stable_surfaces.get("post"), "stable shell post")
    if stable_surfaces["pre"] != stable_surfaces["post"] or stable_surfaces["post"] != frames[-1].get("surfaces"):
        raise ValueError("transition observations require stable shell surface consistency")
    if observations.get("prior_body_removed") is not True:
        raise ValueError("transition observations require prior body removal")
    loading_frames = [frame for frame in frames if frame.get("loading") is True]
    if observations.get("loading_observed") is not True or not loading_frames:
        raise ValueError("transition observations require observed loading state")
    if any(frame.get("prior_body_visible") is not False for frame in loading_frames):
        raise ValueError("transition observations require prior body absence while loading")
    if not observations.get("target_page"):
        raise ValueError("transition observations require target page")
    if observations.get("final_agreement") is not True:
        raise ValueError("transition observations require target body agreement")
    runtime = manifest.get("runtime") or {}
    if not runtime.get("viewport") or not isinstance(runtime.get("interval_ms"), int):
        raise ValueError("transition manifest requires viewport and interval")
    command = manifest.get("command") or {}
    if command.get("exit_code") != 0 or command.get("teardown") != "complete":
        raise ValueError("transition manifest requires successful command teardown")
    source_revision = manifest.get("source_revision")
    if not isinstance(source_revision, str) or len(source_revision) != 40:
        raise ValueError("transition manifest requires full source revision")
    source_files = manifest.get("source_files")
    if not isinstance(source_files, dict) or not source_files or any(
        not isinstance(path, str)
        or not isinstance(digest, str)
        or len(digest) != 64
        for path, digest in source_files.items()
    ):
        raise ValueError("transition manifest requires source files and SHA-256 digests")
    source_paths = tuple(source_files)
    if manifest.get("source_revision_kind") != "git-head-base":
        raise ValueError("transition manifest requires git-head-base revision kind")
    if source_revision != current_git_revision():
        raise ValueError("transition manifest source revision does not match current Git HEAD")
    if source_files != source_file_digests(source_paths):
        raise ValueError("transition manifest source file digests do not match current files")
    if manifest.get("source_worktree_dirty") != git_worktree_dirty(source_paths):
        raise ValueError("transition manifest working-tree state does not match current files")
    if not isinstance(manifest.get("evidence_command"), list) or not manifest["evidence_command"]:
        raise ValueError("transition manifest requires evidence command")
    if not runtime.get("host_architecture") or not runtime.get("browser_user_agent"):
        raise ValueError("transition manifest requires runtime identity")


def current_git_revision() -> str:
    return subprocess.run(
        ["git", "rev-parse", "HEAD"],
        check=True,
        capture_output=True,
        text=True,
    ).stdout.strip()


def git_worktree_dirty(paths: tuple[str, ...]) -> bool:
    result = subprocess.run(
        ["git", "status", "--porcelain", "--", *paths],
        check=True,
        capture_output=True,
        text=True,
    )
    return bool(result.stdout.strip())


def source_file_digests(paths: tuple[str, ...]) -> dict[str, str]:
    for path in paths:
        candidate = Path(path)
        if candidate.is_absolute() or ".." in candidate.parts:
            raise ValueError("source file paths must be repository-relative")
    return {path: hashlib.sha256(Path(path).read_bytes()).hexdigest() for path in paths}


async def cdp_transition(ws_url: str, base_url: str, from_page: str, to_page: str,
                         screenshot: Path, snapshot: Path, interval_ms: int, frame_count: int,
                         admin_id: str, admin_password: str) -> dict[str, Any]:
    async with websockets.connect(ws_url, max_size=16 * 1024 * 1024) as ws:
        cdp = CdpClient(ws)
        await cdp.call("Page.enable")
        await cdp.call("Runtime.enable")
        await cdp.call("Emulation.setDeviceMetricsOverride", {"width": 1440, "height": 1400, "deviceScaleFactor": 1, "mobile": False})
        await cdp.call("Page.navigate", {"url": base_url})
        await asyncio.sleep(1)
        login_expression = """(async () => {
          const body = new URLSearchParams({admin_id: %s, admin_password: %s});
          const response = await fetch('/api/auth/login', {method:'POST', credentials:'include', headers:{'Content-Type':'application/x-www-form-urlencoded'}, body});
          return {status: response.status, location: response.url};
        })()""" % (json.dumps(admin_id), json.dumps(admin_password))
        await cdp.call("Runtime.evaluate", {"expression": login_expression, "awaitPromise": True, "returnByValue": True})
        await cdp.call("Page.navigate", {"url": f"{base_url}/?page={urllib.parse.quote(from_page)}"})
        deadline = time.time() + 45
        while time.time() < deadline:
            check = await cdp.call("Runtime.evaluate", {"expression": "document.body ? document.body.innerText : ''", "returnByValue": True})
            if from_page in (check.get("result", {}).get("value", "") or ""):
                break
            await asyncio.sleep(.5)
        install = """(() => {
          window.__evidenceEvents=[];
          window.__evidenceDocumentMarker ||= crypto.randomUUID();
          window.__evidencePriorBodyText = (document.querySelector('[data-testid=stMain]') || document.body)?.innerText || '';
          const sidebar=document.querySelector('[data-testid=stSidebar]');
          if (sidebar) sidebar.dataset.evidenceMarker ||= crypto.randomUUID();
          window.__evidenceSurface = (selector) => { const element=document.querySelector(selector); if (!element) return null; const style=getComputedStyle(element); const rect=element.getBoundingClientRect(); return {backgroundImage:style.backgroundImage, backgroundColor:style.backgroundColor, visibility:style.visibility, display:style.display, opacity:style.opacity, rendered:rect.width>0 && rect.height>0 && style.visibility!=='hidden' && style.display!=='none' && Number(style.opacity)>0}; };
          for (const name of ['pushState','replaceState']) {
            const old=history[name];
            history[name]=function(){window.__evidenceEvents.push({type:name,url:String(arguments[2]||location.href)}); return old.apply(this,arguments)};
          }
        })()"""
        await cdp.call("Runtime.evaluate", {"expression": install})
        cdp.events.clear()
        observations: list[dict[str, Any]] = []
        click = """(() => { const labels=[...document.querySelectorAll('label')]; const target=labels.find(x=>x.innerText.trim()===%s); if(target){target.click(); return true;} const input=[...document.querySelectorAll('input')].find(x=>x.value===%s); if(input){input.click(); return true;} return false; })()""" % (json.dumps(to_page), json.dumps(to_page))
        deadline = time.time() + 45
        index = 0
        while time.time() < deadline and (index < frame_count or not observations or observations[-1].get("selected_menu") != to_page):
            expression = """(() => { const body=document.body; const text=body?.innerText||''; const main=document.querySelector('[data-testid=stMain]')||body; const mainText=main.innerText||''; const sidebar=document.querySelector('[data-testid=stSidebar]'); const selectedInput=document.querySelector('[data-testid=stRadio] input:checked'); const selected=selectedInput?.closest('label')?.innerText?.trim() || document.querySelector('[aria-current=page]')?.textContent?.trim() || ''; const prior=window.__evidencePriorBodyText||''; return {at_ms:%d, selected_menu:selected, body_text:mainText, prior_body_visible:Boolean(prior)&&mainText.includes(prior), document_marker:window.__evidenceDocumentMarker||'', sidebar_marker:sidebar?.dataset.evidenceMarker||'', background:getComputedStyle(document.body).backgroundColor, surfaces:{stAppViewContainer:window.__evidenceSurface('[data-testid=stAppViewContainer]'), stSidebar:window.__evidenceSurface('[data-testid=stSidebar]')}, loading:/불러오는 중/.test(text)}; })()""" % (index * interval_ms)
            result = await cdp.call("Runtime.evaluate", {"expression": expression, "returnByValue": True})
            frame = result.get("result", {}).get("value", {})
            frame["body_text_hash"] = hashlib.sha256(frame.get("body_text", "").encode()).hexdigest()
            frame.pop("body_text", None)
            observations.append(frame)
            if index == 0:
                await cdp.call("Runtime.evaluate", {"expression": click, "returnByValue": True})
            index += 1
            await asyncio.sleep(interval_ms / 1000)
        final = await cdp.call("Runtime.evaluate", {"expression": "(() => { const main=document.querySelector('[data-testid=stMain]')||document.body; const sidebar=document.querySelector('[data-testid=stSidebar]'); const selectedInput=document.querySelector('[data-testid=stRadio] input:checked'); return {url:location.href, body:main.innerText||'', selected:selectedInput?.closest('label')?.innerText?.trim() || document.querySelector('[aria-current=page]')?.textContent?.trim() || '', document_marker:window.__evidenceDocumentMarker||'', sidebar_marker:sidebar?.dataset.evidenceMarker||'', background:getComputedStyle(document.body).backgroundColor, surfaces:{stAppViewContainer:window.__evidenceSurface('[data-testid=stAppViewContainer]'), stSidebar:window.__evidenceSurface('[data-testid=stSidebar]')}, history_events:window.__evidenceEvents||[], browser_user_agent:navigator.userAgent}; })()", "returnByValue": True})
        data = final.get("result", {}).get("value", {})
        html = await cdp.call("Runtime.evaluate", {"expression": "document.documentElement.outerHTML", "returnByValue": True})
        html_text = html.get("result", {}).get("value", "")
        snapshot.write_text(redact_secrets(html_text, (admin_id, admin_password)), encoding="utf-8")
        image = await cdp.call("Page.captureScreenshot", {"format":"png", "fromSurface":True, "captureBeyondViewport":True})
        screenshot.write_bytes(base64.b64decode(image["data"]))
        document_events = [{"type": event.get("method"), "params": event.get("params", {})} for event in cdp.events if event.get("method") in ("Page.frameNavigated", "Page.loadEventFired")]
        pre = observations[0] if observations else {}
        loading_frames = [frame for frame in observations if frame.get("loading")]
        return {"frames": observations, "document_navigation_events": document_events, "history_events": data.get("history_events", []), "stable_shell": {"document_marker": {"pre": pre.get("document_marker"), "post": data.get("document_marker")}, "sidebar_marker": {"pre": pre.get("sidebar_marker"), "post": data.get("sidebar_marker")}, "background": {"pre": pre.get("background"), "post": data.get("background")}, "surfaces": {"pre": pre.get("surfaces"), "post": data.get("surfaces")}}, "loading_observed": bool(loading_frames), "prior_body_removed": bool(loading_frames) and all(frame.get("prior_body_visible") is False for frame in loading_frames), "target_page": to_page, "final_agreement": data.get("selected") == to_page and to_page in data.get("body", ""), "final_url": data.get("url", ""), "browser_user_agent": data.get("browser_user_agent", "")}


def transition_capture(chrome: str, base_url: str, from_page: str, to_page: str, output_dir: Path, devtools_port: int,
                       interval_ms: int, frame_count: int,
                       admin_id: str, admin_password: str) -> dict[str, Any]:
    profile_dir = output_dir / "chrome-profile-transition"
    screenshot, snapshot = output_dir / "transition.png", output_dir / "transition.dom.html"
    chrome_cmd = [chrome, "--headless=new", "--no-sandbox", "--disable-dev-shm-usage", "--ignore-certificate-errors", f"--remote-debugging-port={devtools_port}", f"--user-data-dir={profile_dir}", "--window-size=1440,1400", "about:blank"]
    proc = subprocess.Popen(chrome_cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    try:
        ws_url = wait_for_devtools(devtools_port)
        return asyncio.run(cdp_transition(ws_url, base_url, from_page, to_page, screenshot, snapshot, interval_ms, frame_count, admin_id, admin_password))
    finally:
        if proc.poll() is None:
            proc.terminate()
            try: proc.wait(timeout=8)
            except subprocess.TimeoutExpired: proc.kill(); proc.wait(timeout=5)
        remove_tree_with_retries(profile_dir)


def main() -> int:
    args = parse_args()
    chrome = require_executable(args.chrome, "Chrome")
    streamlit = None if args.mode == "transition" else require_executable(args.streamlit, "Streamlit")
    run_id = datetime.now().strftime("run-%Y%m%d-%H%M%S")
    output_dir = Path(args.output_dir) if args.output_dir else DEFAULT_OUTPUT_ROOT / run_id
    output_dir.mkdir(parents=True, exist_ok=True)
    db_path = output_dir / "fixture.sqlite3"
    fixture = seed_fixture_db(db_path)

    env = os.environ.copy()
    env["JARCA_LEDGER_DB_PATH"] = str(db_path.resolve())
    env["STREAMLIT_BROWSER_GATHER_USAGE_STATS"] = "false"
    transition_secrets: tuple[str, ...] = ()
    if args.mode == "transition":
        if not args.cert_file or not args.key_file:
            raise SystemExit("transition mode requires --cert-file and --key-file")
        env["JARCA_DASHBOARD_ADMIN_ID"] = secrets.token_urlsafe(18)
        env["JARCA_DASHBOARD_ADMIN_PASSWORD"] = secrets.token_urlsafe(32)
        env["JARCA_DASHBOARD_SIGNING_SECRET"] = secrets.token_hex(32)
        transition_secrets = (env["JARCA_DASHBOARD_ADMIN_ID"], env["JARCA_DASHBOARD_ADMIN_PASSWORD"], env["JARCA_DASHBOARD_SIGNING_SECRET"])
    scheme = "https" if args.mode == "transition" else "http"
    base_url = f"{scheme}://{args.host}:{args.port}"
    if args.mode == "transition":
        server_cmd = [sys.executable, "-m", "uvicorn", "dashboard.server:app", "--host", args.host, "--port", str(args.port), "--ssl-certfile", args.cert_file, "--ssl-keyfile", args.key_file]
    else:
        server_cmd = [
            streamlit, "run", "dashboard/app.py", "--server.address", args.host,
            "--server.port", str(args.port), "--server.headless", "true",
            "--browser.gatherUsageStats", "false",
        ]

    server = subprocess.Popen(server_cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env, start_new_session=True)
    manifest: dict[str, Any] = {
        "success": False,
        "mode": args.mode,
        "base_url": base_url,
        "output_dir": str(output_dir.resolve()),
        "fixture": {"db_path": fixture["db_path"]},
        "chrome": chrome,
        "streamlit": streamlit,
        "pages": [],
        "server_command": server_cmd,
    }
    error: str | None = None
    try:
        wait_for_http(base_url, ssl_context=ssl._create_unverified_context() if args.mode == "transition" else None)
        if args.mode == "transition":
            observations = transition_capture(chrome, base_url, args.from_page, args.to_page, output_dir, 19222, args.interval_ms, args.frames, env["JARCA_DASHBOARD_ADMIN_ID"], env["JARCA_DASHBOARD_ADMIN_PASSWORD"])
            source_paths = (
                "dashboard/app.py",
                "dashboard/common.py",
                "scripts/streamlit_browser_evidence.py",
            )
            manifest.update(
                {
                    "observations": observations,
                    "runtime": {
                        "viewport": {"width": 1440, "height": 1400},
                        "interval_ms": args.interval_ms,
                        "frame_count": args.frames,
                        "host_architecture": platform.machine(),
                        "python_version": platform.python_version(),
                        "browser_user_agent": observations.get("browser_user_agent", ""),
                    },
                    "source_revision": current_git_revision(),
                    "source_revision_kind": "git-head-base",
                    "source_worktree_dirty": git_worktree_dirty(source_paths),
                    "source_files": source_file_digests(source_paths),
                    "evidence_command": ["scripts/streamlit_browser_evidence.py", *sys.argv[1:]],
                    "limitations": ["TLS uses caller-provided local certificate"],
                }
            )
        else:
            for index, page in enumerate(args.pages or DEFAULT_PAGES):
                manifest["pages"].append(capture_page(chrome, base_url, page, output_dir, devtools_port=19222 + index))
    except Exception as exc:  # noqa: BLE001 - failure is recorded in redacted evidence
        error = str(exc)
    finally:
        if not args.keep_server and server.poll() is None:
            os.killpg(server.pid, signal.SIGTERM)
            try:
                server.wait(timeout=10)
            except subprocess.TimeoutExpired:
                os.killpg(server.pid, signal.SIGKILL)
                server.wait(timeout=5)
        returncode = server.poll()
        teardown = "complete" if not args.keep_server and returncode is not None else "incomplete"
        effective_exit = 0 if returncode in (0, -signal.SIGTERM) else returncode
        manifest["server_log_tail"] = ""
        manifest["command"] = {"exit_code": effective_exit, "teardown": teardown}
        if error:
            manifest["error"] = error
        if args.mode == "transition" and error is None and teardown == "complete" and effective_exit == 0:
            try:
                manifest["success"] = True
                validate_transition_manifest(manifest, transition_secrets)
            except Exception as exc:  # noqa: BLE001 - preserve failed manifest
                manifest["success"] = False
                manifest["error"] = str(exc)
        elif args.mode == "capture" and error is None and (
            (args.keep_server and returncode is None)
            or (teardown == "complete" and effective_exit == 0)
        ):
            manifest["success"] = True
        safe_manifest = redact_secrets(manifest, transition_secrets)
        manifest_path = output_dir / "manifest.json"
        temporary_path = output_dir / ".manifest.json.tmp"
        temporary_path.write_text(json.dumps(safe_manifest, ensure_ascii=False, indent=2), encoding="utf-8")
        temporary_path.replace(manifest_path)
    print(json.dumps(safe_manifest, ensure_ascii=False, indent=2))
    return 0 if manifest.get("success") is True else 1


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