liaohch3--claude-tap/scripts/check_coverage.py

#!/usr/bin/env python3
"""Enforce project and incremental coverage targets.

Python coverage is read from a coverage.py JSON report. Viewer frontend coverage
is measured with Chromium V8 precise coverage against the cross-client viewer
contract traces. The frontend incremental metric is function-oriented: changed
viewer JavaScript functions must be exercised by V8 coverage.
"""

from __future__ import annotations

import argparse
import importlib.util
import json
import re
import subprocess
import sys
import tempfile
import tomllib
from dataclasses import dataclass
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONFIG = REPO_ROOT / "pyproject.toml"
DEFAULT_THRESHOLDS = {
    "python_total_min": 65.0,
    "python_diff_min": 80.0,
    "viewer_js_function_min": 50.0,
    "viewer_js_diff_min": 80.0,
    "viewer_css_selector_min": 65.0,
    "viewer_css_diff_min": 80.0,
}
VIEWER_HTML_SOURCE = "claude_tap/viewer.html"
VIEWER_LEGACY_JS_SOURCE = "claude_tap/viewer_assets/viewer.js"
VIEWER_JS_SOURCES = (
    "claude_tap/viewer_assets/state.js",
    "claude_tap/viewer_assets/responses.js",
    "claude_tap/viewer_assets/lazy_loading.js",
    "claude_tap/viewer_assets/i18n_ui.js",
    "claude_tap/viewer_assets/live_bootstrap.js",
    "claude_tap/viewer_assets/filters_search.js",
    "claude_tap/viewer_assets/sidebar.js",
    "claude_tap/viewer_assets/detail_trace.js",
    "claude_tap/viewer_assets/renderers.js",
    "claude_tap/viewer_assets/sections_json.js",
    "claude_tap/viewer_assets/diff.js",
    "claude_tap/viewer_assets/utilities_mobile.js",
)
VIEWER_CSS_SOURCE = "claude_tap/viewer_assets/viewer.css"
VIEWER_DIFF_PATHS = ["claude_tap/*.py", VIEWER_HTML_SOURCE, "claude_tap/viewer_assets/*"]
VIEWER_STYLE_TEMPLATE_ANCHOR = "<!-- CLAUDE_TAP_VIEWER_STYLE -->"
VIEWER_SCRIPT_TEMPLATE_ANCHOR = "<!-- CLAUDE_TAP_VIEWER_SCRIPT -->"


@dataclass(frozen=True)
class CheckResult:
    name: str
    percent: float | None
    minimum: float
    passed: bool
    detail: str


def _run_git_diff(base: str, paths: list[str]) -> str:
    cmd = ["git", "diff", "--unified=0", f"{base}...HEAD", "--", *paths]
    return subprocess.check_output(cmd, cwd=REPO_ROOT, text=True)


def changed_lines_from_diff(diff_text: str) -> dict[str, set[int]]:
    changed: dict[str, set[int]] = {}
    current_file: str | None = None
    new_line: int | None = None
    hunk_re = re.compile(r"@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@")

    for line in diff_text.splitlines():
        if line.startswith("diff --git "):
            current_file = None
            new_line = None
            continue
        if line.startswith("+++ b/"):
            current_file = line[len("+++ b/") :]
            changed.setdefault(current_file, set())
            continue
        if line.startswith("@@ "):
            match = hunk_re.search(line)
            if not match:
                new_line = None
                continue
            new_line = int(match.group(1))
            continue
        if current_file is None or new_line is None:
            continue
        if line.startswith("+") and not line.startswith("+++"):
            changed[current_file].add(new_line)
            new_line += 1
        elif line.startswith("-") and not line.startswith("---"):
            continue
        else:
            new_line += 1

    return {path: lines for path, lines in changed.items() if lines}


def _tag_content(source: str, tag: str) -> str | None:
    start_tag = f"<{tag}>"
    end_tag = f"</{tag}>"
    start = source.find(start_tag)
    if start < 0:
        return None
    content_start = start + len(start_tag)
    end = source.find(end_tag, content_start)
    if end < 0:
        return None
    return source[content_start:end].strip("\n")


def _replace_tag_block(source: str, tag: str, replacement: str) -> str | None:
    start_tag = f"<{tag}>"
    end_tag = f"</{tag}>"
    start = source.find(start_tag)
    if start < 0:
        return None
    end = source.find(end_tag, start + len(start_tag))
    if end < 0:
        return None
    return source[:start] + replacement + source[end + len(end_tag) :]


def _base_file_text(base: str, path: str) -> str | None:
    try:
        return subprocess.check_output(["git", "show", f"{base}:{path}"], cwd=REPO_ROOT, text=True)
    except subprocess.CalledProcessError:
        return None


def _filter_pure_viewer_asset_split(changed_lines: dict[str, set[int]], base: str) -> dict[str, set[int]]:
    """Ignore asset files that are exact extractions from the base monolithic viewer."""
    js_changed = [source for source in (VIEWER_LEGACY_JS_SOURCE, *VIEWER_JS_SOURCES) if source in changed_lines]
    if not js_changed and VIEWER_CSS_SOURCE not in changed_lines and VIEWER_HTML_SOURCE not in changed_lines:
        return changed_lines

    base_viewer = _base_file_text(base, VIEWER_HTML_SOURCE)
    if base_viewer is None:
        return changed_lines

    filtered = dict(changed_lines)
    expected_js = _tag_content(base_viewer, "script")
    expected_css = _tag_content(base_viewer, "style")

    js_exact = False
    if js_changed and expected_js is not None:
        try:
            if (REPO_ROOT / VIEWER_LEGACY_JS_SOURCE).exists():
                current_js = (REPO_ROOT / VIEWER_LEGACY_JS_SOURCE).read_text(encoding="utf-8").strip("\n")
            else:
                current_js = "".join((REPO_ROOT / source).read_text(encoding="utf-8") for source in VIEWER_JS_SOURCES)
                current_js = current_js.strip("\n")
        except OSError:
            current_js = None
        js_exact = current_js == expected_js
        if js_exact:
            for source in (VIEWER_LEGACY_JS_SOURCE, *VIEWER_JS_SOURCES):
                filtered.pop(source, None)

    css_exact = False
    if VIEWER_CSS_SOURCE in filtered and expected_css is not None:
        try:
            current_css = (REPO_ROOT / VIEWER_CSS_SOURCE).read_text(encoding="utf-8").strip("\n")
        except OSError:
            current_css = None
        css_exact = current_css == expected_css
        if css_exact:
            filtered.pop(VIEWER_CSS_SOURCE, None)

    if (
        VIEWER_HTML_SOURCE in filtered
        and (js_exact or not js_changed)
        and (css_exact or VIEWER_CSS_SOURCE not in changed_lines)
    ):
        expected_template = _replace_tag_block(base_viewer, "style", VIEWER_STYLE_TEMPLATE_ANCHOR)
        if expected_template is not None:
            expected_template = _replace_tag_block(expected_template, "script", VIEWER_SCRIPT_TEMPLATE_ANCHOR)
        try:
            current_template = (REPO_ROOT / VIEWER_HTML_SOURCE).read_text(encoding="utf-8")
        except OSError:
            current_template = None
        if (
            expected_template is not None
            and current_template is not None
            and current_template.strip() == expected_template.strip()
        ):
            filtered.pop(VIEWER_HTML_SOURCE, None)
    return filtered


def load_thresholds(config_path: Path = DEFAULT_CONFIG) -> dict[str, float]:
    data = tomllib.loads(config_path.read_text(encoding="utf-8"))
    configured = data.get("tool", {}).get("claude_tap", {}).get("coverage", {})
    thresholds = dict(DEFAULT_THRESHOLDS)
    for key in thresholds:
        if key in configured:
            thresholds[key] = float(configured[key])
    return thresholds


def check_python_coverage(
    coverage_json_path: Path,
    changed_lines: dict[str, set[int]],
    total_min: float,
    diff_min: float,
) -> list[CheckResult]:
    coverage = json.loads(coverage_json_path.read_text(encoding="utf-8"))
    total_percent = float(coverage["totals"]["percent_covered"])
    results = [
        CheckResult(
            name="python_total",
            percent=total_percent,
            minimum=total_min,
            passed=total_percent >= total_min,
            detail=f"coverage.py total {total_percent:.2f}% >= {total_min:.2f}%",
        )
    ]

    executable_changed = 0
    covered_changed = 0
    files = coverage.get("files", {})
    for path, changed in changed_lines.items():
        if not path.startswith("claude_tap/") or not path.endswith(".py"):
            continue
        file_cov = files.get(path)
        if not file_cov:
            executable_changed += len(changed)
            continue
        executed = set(file_cov.get("executed_lines", []))
        missing = set(file_cov.get("missing_lines", []))
        executable = executed | missing
        relevant = changed & executable
        executable_changed += len(relevant)
        covered_changed += len(relevant & executed)

    if executable_changed == 0:
        results.append(
            CheckResult(
                name="python_diff",
                percent=None,
                minimum=diff_min,
                passed=True,
                detail="no changed executable Python package lines",
            )
        )
        return results

    diff_percent = covered_changed / executable_changed * 100
    results.append(
        CheckResult(
            name="python_diff",
            percent=diff_percent,
            minimum=diff_min,
            passed=diff_percent >= diff_min,
            detail=f"{covered_changed}/{executable_changed} changed executable Python lines covered",
        )
    )
    return results


def js_function_ranges(source: str) -> dict[str, tuple[int, int]]:
    lines = source.splitlines()
    ranges: dict[str, tuple[int, int]] = {}
    fn_re = re.compile(r"\bfunction\s+([A-Za-z_$][\w$]*)\s*\(")

    line_no = 1
    while line_no <= len(lines):
        line = lines[line_no - 1]
        match = fn_re.search(line)
        if not match:
            line_no += 1
            continue

        name = match.group(1)
        depth = 0
        seen_open = False
        end_line = line_no
        scan = line_no
        while scan <= len(lines):
            for char in lines[scan - 1]:
                if char == "{":
                    depth += 1
                    seen_open = True
                elif char == "}":
                    depth -= 1
                    if seen_open and depth <= 0:
                        end_line = scan
                        break
            if seen_open and depth <= 0:
                break
            scan += 1

        ranges[name] = (line_no, end_line)
        line_no = max(end_line + 1, line_no + 1)

    return ranges


def _changed_lines_for_source(source_path: Path, changed_lines: dict[str, set[int]], fallback_key: str) -> set[int]:
    keys = []
    if source_path.name == "viewer.js":
        keys.append(VIEWER_LEGACY_JS_SOURCE)
    elif source_path.name == "viewer.css":
        keys.append(VIEWER_CSS_SOURCE)
    try:
        keys.insert(0, source_path.resolve().relative_to(REPO_ROOT).as_posix())
    except ValueError:
        pass
    if source_path.name not in {
        Path(source).name for source in (*VIEWER_JS_SOURCES, VIEWER_LEGACY_JS_SOURCE, VIEWER_CSS_SOURCE)
    }:
        keys.append(fallback_key)
    for key in keys:
        changed = changed_lines.get(key)
        if changed:
            return changed
    return set()


def changed_viewer_functions(viewer_js: Path | tuple[Path, ...], changed_lines: dict[str, set[int]]) -> set[str]:
    functions: set[str] = set()
    for source_path in viewer_js if isinstance(viewer_js, tuple) else (viewer_js,):
        changed = _changed_lines_for_source(source_path, changed_lines, VIEWER_HTML_SOURCE)
        if not changed:
            continue
        ranges = js_function_ranges(source_path.read_text(encoding="utf-8"))
        for name, (start, end) in ranges.items():
            if any(start <= line <= end for line in changed):
                functions.add(name)
    return functions


def _split_selectors(selector_text: str) -> list[str]:
    return [part.strip() for part in selector_text.split(",") if part.strip()]


def _is_queryable_selector(selector: str) -> bool:
    return not re.search(
        r"::|:(hover|active|focus|focus-visible|focus-within|visited|target|checked|disabled|enabled|placeholder-shown)\b",
        selector,
    )


def _style_block_ranges(source: str) -> list[tuple[int, int]]:
    ranges: list[tuple[int, int]] = []
    start: int | None = None
    for idx, line in enumerate(source.splitlines(), start=1):
        if "<style>" in line:
            start = idx + 1
            continue
        if "</style>" in line and start is not None:
            ranges.append((start, idx - 1))
            start = None
    if not ranges:
        ranges.append((1, len(source.splitlines())))
    return ranges


def css_selector_ranges(source: str) -> dict[str, list[tuple[int, int]]]:
    lines = source.splitlines()
    ranges: dict[str, list[tuple[int, int]]] = {}

    for style_start, style_end in _style_block_ranges(source):
        line_no = style_start
        while line_no <= style_end:
            line = lines[line_no - 1]
            if "{" not in line:
                line_no += 1
                continue
            selector_text = line.split("{", 1)[0].strip()
            if not selector_text or selector_text.startswith("@"):
                line_no += 1
                continue

            depth = 0
            end_line = line_no
            for scan in range(line_no, style_end + 1):
                for char in lines[scan - 1]:
                    if char == "{":
                        depth += 1
                    elif char == "}":
                        depth -= 1
                if depth <= 0:
                    end_line = scan
                    break

            for selector in _split_selectors(selector_text):
                if _is_queryable_selector(selector):
                    ranges.setdefault(selector, []).append((line_no, end_line))
            line_no = max(end_line + 1, line_no + 1)

    return ranges


def changed_viewer_css_selectors(viewer_css: Path, changed_lines: dict[str, set[int]]) -> set[str]:
    changed = _changed_lines_for_source(viewer_css, changed_lines, VIEWER_HTML_SOURCE)
    if not changed:
        return set()
    ranges = css_selector_ranges(viewer_css.read_text(encoding="utf-8"))
    selectors: set[str] = set()
    for selector, selector_ranges in ranges.items():
        if any(start <= line <= end for start, end in selector_ranges for line in changed):
            selectors.add(selector)
    return selectors


def _main_viewer_script(coverage: dict[str, Any], suffix: str) -> dict[str, Any]:
    candidates = [
        script
        for script in coverage["result"]
        if script.get("url", "").endswith(suffix) and len(script.get("functions", [])) > 50
    ]
    if not candidates:
        raise RuntimeError("Could not find viewer.html main script in V8 coverage output")
    return max(candidates, key=lambda script: len(script.get("functions", [])))


def _is_top_level_wrapper(function: dict[str, Any], script_end: int) -> bool:
    if function.get("functionName"):
        return False
    ranges = function.get("ranges", [])
    if not ranges:
        return False
    widest = max((item.get("endOffset", 0) - item.get("startOffset", 0) for item in ranges), default=0)
    return script_end > 0 and widest >= script_end * 0.8


def _viewer_script_functions(script: dict[str, Any]) -> list[dict[str, Any]]:
    all_ranges = [item for function in script["functions"] for item in function.get("ranges", [])]
    script_end = max((item.get("endOffset", 0) for item in all_ranges), default=0)
    return [function for function in script["functions"] if not _is_top_level_wrapper(function, script_end)]


def _is_function_covered(function: dict[str, Any]) -> bool:
    return any(item.get("count", 0) > 0 for item in function.get("ranges", []))


def _load_viewer_contract_helpers() -> tuple[Any, Any, Any]:
    contracts_path = REPO_ROOT / "tests" / "test_viewer_contracts.py"
    spec = importlib.util.spec_from_file_location("viewer_contracts_for_coverage", contracts_path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"Could not load viewer contract helpers from {contracts_path}")
    contracts = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = contracts
    spec.loader.exec_module(contracts)
    return contracts._contract_cases, contracts._generate_case_html, contracts._compact_contract_records


def collect_viewer_js_coverage() -> tuple[float, set[str], int, int]:
    try:
        from playwright.sync_api import sync_playwright
    except ImportError as exc:  # pragma: no cover - exercised in dependency-free environments
        raise RuntimeError("Playwright is required for viewer JS coverage") from exc

    _contract_cases, _generate_case_html, _compact_contract_records = _load_viewer_contract_helpers()

    with tempfile.TemporaryDirectory() as tmp:
        tmp_path = Path(tmp)
        html_path = _generate_case_html(
            tmp_path,
            "v8_coverage",
            tuple(record for case in _contract_cases() for record in case.records),
        )
        empty_html_path = _generate_case_html(tmp_path, "empty_coverage", ())
        compact_html_path = tmp_path / "compact_coverage.html"
        compact_bundle_path = tmp_path / "compact_coverage.ctap.json"
        remote_html_path = tmp_path / "remote_coverage.html"
        from claude_tap.compact_trace import build_compact_trace_bundle
        from claude_tap.viewer import (
            _extract_metadata_from_record,
            _generate_html_viewer_from_compact_bundle,
            _generate_html_viewer_from_metadata,
        )

        compact_bundle = build_compact_trace_bundle(list(_compact_contract_records()))
        _generate_html_viewer_from_compact_bundle(
            compact_bundle,
            compact_html_path,
            display_trace_path=compact_bundle_path,
            display_html_path=compact_html_path,
        )
        compact_bundle_path.write_text(
            json.dumps(compact_bundle, ensure_ascii=False, separators=(",", ":")),
            encoding="utf-8",
        )
        remote_record = next(record for case in _contract_cases() for record in case.records)
        remote_api_url = "https://coverage.local/api/records"
        _generate_html_viewer_from_metadata(
            [_extract_metadata_from_record(remote_record)],
            remote_html_path,
            display_trace_path="coverage.ctap.json",
            display_html_path=remote_html_path,
            records_api_path=remote_api_url,
        )
        with sync_playwright() as pw:
            browser = pw.chromium.launch(headless=True)
            page = browser.new_page()
            page.route(
                f"{remote_api_url}**",
                lambda route: route.fulfill(
                    status=200,
                    content_type="application/json",
                    headers={"Access-Control-Allow-Origin": "*"},
                    body=json.dumps(
                        {"session": {"record_count": 1}, "records": [remote_record]},
                        ensure_ascii=False,
                        separators=(",", ":"),
                    ),
                ),
            )
            page.add_init_script("window.__TRACE_SESSION_EXPORTS__ = {jsonl: 'coverage.jsonl', log: 'coverage.log'};")
            session = page.context.new_cdp_session(page)
            session.send("Profiler.enable")
            session.send("Profiler.startPreciseCoverage", {"callCount": True, "detailed": True})
            page.goto(html_path.resolve().as_uri(), timeout=10000)
            page.wait_for_selector(".sidebar-item", timeout=5000)
            page.evaluate(
                """() => {
                  activePaths = new Set(entries.map(getPath));
                  activeTools = null;
                  searchQuery = '';
                  applyFilter(true);
                  setSidebarOrderMode('turn');
                  setSidebarOrderMode('model');
                  setSidebarOrderMode('session');
                  const sidebarItems = sidebarItemsForMode();
                  const sessionGroups = buildSessionGroups(sidebarItems);
                  if (sessionGroups.length) {
                    sessionTextSnippet(sessionGroups[0].userText || 'coverage prompt');
                    finalResponseText(sessionGroups[0].items[0].entry);
                    firstUserInputInfo(sessionGroups[0].items[0].entry);
                  }
                  const continuationEntry = {
                    request_id: 'coverage_continuation',
                    turn: '999.2',
                    request: {
                      body: {
                        previous_response_id: 'resp_coverage',
                        input: [{ type: 'function_call_output', output: 'ok' }]
                      }
                    },
                    response: {
                      body: {
                        id: 'resp_coverage_next',
                        previous_response_id: 'resp_coverage',
                        output: [
                          {
                            type: 'message',
                            role: 'assistant',
                            content: [{ type: 'output_text', text: 'coverage done' }]
                          }
                        ]
                      }
                    }
                  };
                  isContinuationWithoutUserInput(continuationEntry);
                  sessionKeyForEntry(continuationEntry, { key: 'coverage', userText: '', responseText: '', items: [] });
                  if (entries.length) {
                    sessionTurnDiscriminator(entries[0]);
                    sessionKeyForEntry(entries[0], null);
                    matchSearch(entries[0], '1');
                    const originalPrompt = window.prompt;
                    window.prompt = () => '1';
                    promptJumpToTurn();
                    window.prompt = originalPrompt;
                    _buildDiffTargetOptions(Math.min(1, filtered.length - 1));
                    if (filtered.length > 1) showDiffForIdx(1, null, 0);
                  }
                  const stubEntry = buildStubEntry({
                    turn: '2.2',
                    transport: 'websocket',
                    method: 'WEBSOCKET',
                    path: '/v1/responses',
                    model: 'gpt-5.5',
                    request_generate: true,
                    response_output_count: 1,
                    output_tokens: 1,
                  }, 0);
                  normalizeDisplayTurns([stubEntry], true);
                  const imageBlock = {
                    type: 'image',
                    source: {
                      type: 'base64',
                      media_type: 'image/png',
                      data: 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+/p9sAAAAASUVORK5CYII='
                    }
                  };
                  imageLookupKey('<session>[Image #1] coverage prompt</session>');
                  isInlineImageUrl('data:image/png;base64,abc');
                  imageSourceFromBlock(imageBlock);
                  imageBlocksForContent([{ type: 'text', text: '[Image #1] coverage prompt' }, imageBlock]);
                  imageSourceKey(imageBlock);
                  buildSessionImageRegistry();
                  naturalTextFromPromptPayload({ prompt: 'coverage prompt' });
                  renderImageElement('data:image/png;base64,abc', 'coverage image');
                  renderImageElementForBlock(imageBlock);
                  document.body.insertAdjacentHTML('beforeend', renderImageBlock(imageBlock, 0, 1, { frameBlocks: true }));
                  renderViewerActions();
                  valueHasReadableEscapes({ cmd: 'printf "coverage\\\\n"' });
                  decodeEscapedTextForView('line1\\\\nline2\\\\t\\\\u4e00');
                  document.body.insertAdjacentHTML(
                    'beforeend',
                    renderToolInput({ cmd: 'printf "coverage\\n"', yield_time_ms: 1000 })
                  );
                  document.querySelector('.tool-input-toggle')?.click();
                  const tooltipTrigger = document.querySelector('.sidebar-group-header') || document.createElement('div');
                  if (!tooltipTrigger.isConnected) document.body.appendChild(tooltipTrigger);
                  tooltipTrigger.dataset.fullUserInput = 'coverage tooltip prompt';
                  sessionTooltip();
                  showSessionTooltip(tooltipTrigger);
                  hideSessionTooltip(tooltipTrigger);
                  formatText('history_delete_done', { count: 1 });
                  updateHistoryDeleteButton();
                  setHistoryDeleteStatus('coverage', 'ok');
                  setHistoryDeleteStatus('', '');
                  deleteSelectedTraceDate();
                  getTargetForGlobalMatch(0);
                  findFilteredIdxByEntryKey(entryStableKey(entries[0]), entries[0]?.request_id || '');
                  openGlobalSearch();
                  globalSearchState.query = 'subagent_type: "';
                  uniqueSearchQueries(['subagent_type', 'subagent_type']);
                  buildGlobalSearchQueries(globalSearchState.query);
                  buildGlobalHighlightQueries(globalSearchState.query);
                  getEntrySearchText(entries[0]);
                  countOneQueryInText('subagent_type: "value"', 'subagent_type');
                  countMatchesInText('subagent_type: "value"', buildGlobalSearchQueries(globalSearchState.query));
                  scheduleGlobalSearchRecalc();
                  flushGlobalSearchRecalc();
                  navigateGlobalSearch(1);
                  revealCurrentSearchMatch();
                  applyGlobalSearchHighlights(0);
                  highlightSearchInContainer(document.querySelector('#detail'), buildGlobalHighlightQueries(globalSearchState.query));
                  findNextSearchMatch('subagent_type: "value"', buildGlobalSearchQueries(globalSearchState.query), 0);
                  closeGlobalSearch();
                  visualNavigate(1);
                  visualNavigate(-1);
                  vsRenderVisible();
                  if (entries.length > 1) compareSidebarModelOrder(entries[0], entries[1]);
                }"""
            )
            page.evaluate(
                """async () => {
                  const liveStatus = document.querySelector('#live-status') || document.createElement('div');
                  liveStatus.id = 'live-status';
                  if (!liveStatus.isConnected) document.body.appendChild(liveStatus);
                  const embeddedRecords = typeof EMBEDDED_TRACE_DATA !== 'undefined'
                    ? EMBEDDED_TRACE_DATA
                    : (typeof EMBEDDED_TRACE_COMPACT_DATA !== 'undefined'
                      ? (materializeCompactTraceBundle(EMBEDDED_TRACE_COMPACT_DATA) || [])
                      : []);
                  const wsRecord = embeddedRecords.find(record => record.transport === 'websocket');
                  if (wsRecord) {
                    expandLiveWebSocketResponseEntries([wsRecord], true);
                    liveRecords = [wsRecord];
                    await onDateChange('live');
                    const OriginalEventSource = window.EventSource;
                    window.EventSource = class {
                      constructor(url) { this.url = url; }
                      close() {}
                    };
                    try {
                      initLiveMode();
                      liveEventSource.onmessage({ data: JSON.stringify(wsRecord) });
                    } finally {
                      window.EventSource = OriginalEventSource;
                    }
                  }
                }"""
            )
            for index in range(page.evaluate("filtered.length")):
                page.evaluate("entryIndex => { detailViewMode = 'default'; selectEntry(entryIndex); }", index)
                page.wait_for_selector("#detail .section", timeout=5000)
                page.evaluate(
                    """(entryIndex) => {
                      const entry = entries[entryIndex];
                      const body = entry.request.body;
                      getMessages(body);
                      getRequestTools(body);
                      extractSystem(body);
                      getUsage(entry);
                      getResponseEvents(entry);
                      getResponseOutput(entry);
                      const jsonSection = Array.from(document.querySelectorAll('#detail .section'))
                        .find(el => el.querySelector('.title')?.textContent === t('section_json'));
                      const jsonToggle = jsonSection?.querySelector('.jt-toggle');
                      if (jsonToggle) {
                        jsonToggle.click();
                        jsonToggle.click();
                      }
                      setDetailViewMode('trace');
                      setTraceFormatMode('json');
                      setTraceFormatMode('yaml');
                      setTraceFormatMode('pretty');
                      renderTracePayload({
                        emptyObject: {},
                        emptyArray: [],
                        nested: { key: 'value' },
                        array: [{ key: 'value' }],
                        multiline: 'line one\\nline two',
                      });
                    }""",
                    index,
                )
            page.goto(compact_html_path.resolve().as_uri(), timeout=10000)
            page.wait_for_selector(".sidebar-item", timeout=5000)
            page.evaluate(
                """(bundleText) => {
                  const parsed = JSON.parse(bundleText);
                  const compactRecords = materializeCompactTraceBundle(parsed);
                  parseTraceText(bundleText);
                  entries = expandWebSocketResponseEntries(compactRecords);
                  applyFilter(true);
                  selectEntry(0);
                  const compactPayload = parsed.records[0];
                  const compactRecord = compactPayload.record;
                  const refPath = parseCompactRefPath(compactPayload.__claude_tap_compact_record__.refs[0].path);
                  const blobRef = compactRecord.request.body.instructions;
                  isCompactBlobRef(blobRef);
                  loadCompactBlobRef(blobRef, parsed.blobs, new Map());
                  materializeCompactRefPath(compactRecord, refPath, parsed.blobs, new Map());
                  materializeCompactRecord(compactPayload, parsed.blobs, new Map());
                  const legacyPayload = {
                    __claude_tap_compact_record__: {
                      version: 1,
                      encoding: 'json-blob-ref',
                    },
                    record: {
                      request: {
                        body: {
                          instructions: blobRef,
                          input: [blobRef],
                        },
                      },
                      response: { body: {} },
                    },
                  };
                  getCompactPath(legacyPayload.record, ['request', 'body', 'instructions']);
                  legacyCompactRefPaths(legacyPayload.record);
                  materializeCompactRecord(legacyPayload, parsed.blobs, new Map());
                }""",
                compact_bundle_path.read_text(encoding="utf-8"),
            )
            page.goto(remote_html_path.resolve().as_uri(), timeout=10000)
            page.wait_for_selector(".sidebar-item", timeout=5000)
            page.evaluate(
                """async () => {
                  const entry = filtered[0];
                  hasEmbeddedRawLines();
                  shouldFetchRemoteEntry(entry);
                  remoteRecordUrl(0);
                  await fetchRemoteEntry(entry);
                  await resolveEntryForDetailAsync(entry);
                  withDisplayFields(entry, entry);
                  selectEntry(0);
                }"""
            )
            page.wait_for_selector("#detail .section", timeout=5000)
            page.goto(empty_html_path.resolve().as_uri(), timeout=10000)
            page.wait_for_selector(".empty-trace-state", timeout=5000)
            page.set_input_files("#file-input", str(compact_bundle_path))
            page.wait_for_selector(".sidebar-item", timeout=5000)
            page.goto(empty_html_path.resolve().as_uri(), timeout=10000)
            page.wait_for_selector(".empty-trace-state", timeout=5000)
            coverage = session.send("Profiler.takePreciseCoverage")
            session.send("Profiler.stopPreciseCoverage")
            session.send("Profiler.disable")
            browser.close()

    main_functions = _viewer_script_functions(_main_viewer_script(coverage, "v8_coverage.html"))
    empty_functions = _viewer_script_functions(_main_viewer_script(coverage, "empty_coverage.html"))
    compact_functions = _viewer_script_functions(_main_viewer_script(coverage, "compact_coverage.html"))
    remote_functions = _viewer_script_functions(_main_viewer_script(coverage, "remote_coverage.html"))
    auxiliary_covered_names = {
        function.get("functionName", "")
        for function in [*empty_functions, *compact_functions, *remote_functions]
        if function.get("functionName") and _is_function_covered(function)
    }
    covered_functions = [
        function
        for function in main_functions
        if _is_function_covered(function) or function.get("functionName", "") in auxiliary_covered_names
    ]
    covered_names = {function.get("functionName", "") for function in covered_functions if function.get("functionName")}
    percent = len(covered_functions) / len(main_functions) * 100 if main_functions else 100.0
    return percent, covered_names, len(covered_functions), len(main_functions)


def collect_viewer_css_coverage() -> tuple[float, set[str], int, int, int]:
    try:
        from playwright.sync_api import sync_playwright
    except ImportError as exc:  # pragma: no cover - exercised in dependency-free environments
        raise RuntimeError("Playwright is required for viewer CSS coverage") from exc

    _contract_cases, _generate_case_html, _ = _load_viewer_contract_helpers()
    collect_css_script = r"""() => {
      const skipped = [];
      const used = new Set();
      const all = [];
      const skipRe = /::|:(hover|active|focus|focus-visible|focus-within|visited|target|checked|disabled|enabled|placeholder-shown)\b/;
      const splitSelectors = text => text.split(',').map(s => s.trim()).filter(Boolean);
      const visit = rules => {
        for (const rule of Array.from(rules || [])) {
          if (rule.type === CSSRule.STYLE_RULE) {
            for (const selector of splitSelectors(rule.selectorText)) {
              if (skipRe.test(selector)) {
                skipped.push(selector);
                continue;
              }
              try {
                all.push(selector);
                if (document.querySelector(selector)) used.add(selector);
              } catch (e) {
                skipped.push(selector);
              }
            }
          } else if (rule.cssRules) {
            visit(rule.cssRules);
          }
        }
      };
      for (const sheet of Array.from(document.styleSheets)) {
        try { visit(sheet.cssRules); } catch (e) {}
      }
      return { used: [...used], all: [...new Set(all)], skipped: [...new Set(skipped)] };
    }"""

    used_selectors: set[str] = set()
    all_selectors: set[str] = set()
    skipped_selectors: set[str] = set()

    def merge(snapshot: dict[str, list[str]]) -> None:
        used_selectors.update(snapshot["used"])
        all_selectors.update(snapshot["all"])
        skipped_selectors.update(snapshot["skipped"])

    with tempfile.TemporaryDirectory() as tmp:
        tmp_path = Path(tmp)
        html_path = _generate_case_html(
            tmp_path,
            "css_usage",
            tuple(record for case in _contract_cases() for record in case.records),
        )
        empty_html_path = _generate_case_html(tmp_path, "empty_css_usage", ())
        with sync_playwright() as pw:
            browser = pw.chromium.launch(headless=True)
            page = browser.new_page(viewport={"width": 1440, "height": 1000})
            page.add_init_script("window.__TRACE_SESSION_EXPORTS__ = {jsonl: 'coverage.jsonl', log: 'coverage.log'};")
            page.goto(html_path.resolve().as_uri(), timeout=10000)
            page.wait_for_selector(".sidebar-item", timeout=5000)
            page.evaluate(
                """() => {
                  activePaths = new Set(entries.map(getPath));
                  activeTools = null;
                  searchQuery = '';
                  applyFilter(true);
                }"""
            )
            page.evaluate(
                """() => {
                  renderViewerActions();
                  const imageBlock = {
                    type: 'image',
                    source: {
                      type: 'base64',
                      media_type: 'image/png',
                      data: 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+/p9sAAAAASUVORK5CYII='
                    }
                  };
                  document.body.insertAdjacentHTML('beforeend', renderImageBlock(imageBlock, 0, 1, { frameBlocks: true }));
                  const tooltipTrigger = document.querySelector('.sidebar-group-header') || document.createElement('div');
                  if (!tooltipTrigger.isConnected) document.body.appendChild(tooltipTrigger);
                  tooltipTrigger.dataset.fullUserInput = 'coverage tooltip prompt';
                  showSessionTooltip(tooltipTrigger);
                  document.body.insertAdjacentHTML(
                    'beforeend',
                    renderToolInput({ cmd: 'printf "coverage\\n"', yield_time_ms: 1000 })
                  );
                  document.querySelector('.tool-input-view')?.classList.add('expanded');
                }"""
            )
            merge(page.evaluate(collect_css_script))
            page.evaluate("setSidebarOrderMode('turn')")
            merge(page.evaluate(collect_css_script))
            page.evaluate("setSidebarOrderMode('model')")
            merge(page.evaluate(collect_css_script))
            page.evaluate(
                """() => {
                  const picker = document.querySelector('#date-picker');
                  if (picker) picker.style.display = 'flex';
                  setHistoryDeleteStatus('coverage ok', 'ok');
                }"""
            )
            merge(page.evaluate(collect_css_script))
            page.evaluate("setHistoryDeleteStatus('coverage warn', 'warn')")
            merge(page.evaluate(collect_css_script))
            page.evaluate("setHistoryDeleteStatus('coverage error', 'error')")
            merge(page.evaluate(collect_css_script))

            for index in range(page.evaluate("filtered.length")):
                page.evaluate("entryIndex => { detailViewMode = 'default'; selectEntry(entryIndex); }", index)
                merge(page.evaluate(collect_css_script))
                page.evaluate("setDetailViewMode('trace')")
                merge(page.evaluate(collect_css_script))
                for mode in ("json", "yaml", "pretty"):
                    page.evaluate("mode => setTraceFormatMode(mode)", mode)
                    merge(page.evaluate(collect_css_script))
                page.evaluate(
                    """() => {
                      document.querySelector('#detail')?.insertAdjacentHTML(
                        'beforeend',
                        renderTracePayload({
                          emptyObject: {},
                          emptyArray: [],
                          nested: { key: 'value' },
                          array: [{ key: 'value' }],
                          multiline: 'line one\\nline two',
                        })
                      );
                    }"""
                )
                merge(page.evaluate(collect_css_script))

            page.evaluate(
                """() => {
                  openGlobalSearch();
                  const input = document.querySelector('#global-search-input');
                  input.value = 'contract';
                  input.dispatchEvent(new Event('input', { bubbles: true }));
                }"""
            )
            merge(page.evaluate(collect_css_script))

            if page.evaluate("filtered.length") > 1:
                page.evaluate("showDiffForIdx(1, null, 0)")
                merge(page.evaluate(collect_css_script))

            page.evaluate(
                """() => {
                  setDetailViewMode('default');
                  const contentBlockEntry = entries.find(entry => entry.request_id === 'req_content_block_boundary_contract');
                  if (contentBlockEntry) renderDetail(contentBlockEntry);
                  document.querySelector('#detail')?.insertAdjacentHTML(
                    'afterbegin',
                    '<div class="continuation-banner"><div class="cb-icon"></div><div class="cb-content"><div class="cb-title"></div><div class="cb-message"></div><div class="cb-meta"><div class="cb-key">id</div><div class="cb-val">resp</div></div></div></div>'
                  );
                }"""
            )
            merge(page.evaluate(collect_css_script))
            page.evaluate("document.documentElement.setAttribute('data-theme', 'dark')")
            merge(page.evaluate(collect_css_script))
            page.set_viewport_size({"width": 390, "height": 900})
            page.evaluate("mobileShowDetail()")
            merge(page.evaluate(collect_css_script))
            page.set_viewport_size({"width": 1440, "height": 900})
            page.goto(
                (
                    f"{html_path.resolve().as_uri()}?embed=1&hideHeader=1&hidePath=1"
                    "&hideHistory=1&hideControls=1&density=compact&theme=light"
                ),
                timeout=10000,
            )
            page.wait_for_selector(".sidebar-item", timeout=5000)
            page.evaluate(
                """() => {
                  const contentBlockEntry = entries.find(entry => entry.request_id === 'req_content_block_boundary_contract');
                  if (contentBlockEntry) renderDetail(contentBlockEntry);
                }"""
            )
            merge(page.evaluate(collect_css_script))
            page.goto(empty_html_path.resolve().as_uri(), timeout=10000)
            page.wait_for_selector(".empty-trace-state", timeout=5000)
            merge(page.evaluate(collect_css_script))
            browser.close()

    percent = len(used_selectors) / len(all_selectors) * 100 if all_selectors else 100.0
    return percent, used_selectors, len(used_selectors), len(all_selectors), len(skipped_selectors)


def check_viewer_js_coverage(
    changed_functions: set[str],
    function_min: float,
    diff_min: float,
) -> list[CheckResult]:
    function_percent, covered_names, covered_count, total_count = collect_viewer_js_coverage()
    results = [
        CheckResult(
            name="viewer_js_functions",
            percent=function_percent,
            minimum=function_min,
            passed=function_percent >= function_min,
            detail=f"{covered_count}/{total_count} V8 functions executed",
        )
    ]

    if not changed_functions:
        results.append(
            CheckResult(
                name="viewer_js_diff",
                percent=None,
                minimum=diff_min,
                passed=True,
                detail="no changed viewer JavaScript functions",
            )
        )
        return results

    covered_changed = changed_functions & covered_names
    diff_percent = len(covered_changed) / len(changed_functions) * 100
    missing = ", ".join(sorted(changed_functions - covered_names)) or "none"
    results.append(
        CheckResult(
            name="viewer_js_diff",
            percent=diff_percent,
            minimum=diff_min,
            passed=diff_percent >= diff_min,
            detail=f"{len(covered_changed)}/{len(changed_functions)} changed JS functions covered; missing: {missing}",
        )
    )
    return results


def check_viewer_css_coverage(
    changed_selectors: set[str],
    selector_min: float,
    diff_min: float,
    coverage: tuple[float, set[str], int, int, int] | None = None,
) -> list[CheckResult]:
    selector_percent, used_selectors, used_count, total_count, skipped_count = coverage or collect_viewer_css_coverage()
    results = [
        CheckResult(
            name="viewer_css_selectors",
            percent=selector_percent,
            minimum=selector_min,
            passed=selector_percent >= selector_min,
            detail=f"{used_count}/{total_count} queryable CSS selectors matched; {skipped_count} state/pseudo selectors skipped",
        )
    ]

    if not changed_selectors:
        results.append(
            CheckResult(
                name="viewer_css_diff",
                percent=None,
                minimum=diff_min,
                passed=True,
                detail="no changed viewer CSS selectors",
            )
        )
        return results

    covered_changed = changed_selectors & used_selectors
    diff_percent = len(covered_changed) / len(changed_selectors) * 100
    missing = ", ".join(sorted(changed_selectors - used_selectors)) or "none"
    results.append(
        CheckResult(
            name="viewer_css_diff",
            percent=diff_percent,
            minimum=diff_min,
            passed=diff_percent >= diff_min,
            detail=f"{len(covered_changed)}/{len(changed_selectors)} changed CSS selectors matched; missing: {missing}",
        )
    )
    return results


def print_results(results: list[CheckResult]) -> None:
    for result in results:
        status = "PASS" if result.passed else "FAIL"
        if result.percent is None:
            print(f"{status} {result.name}: {result.detail} (target {result.minimum:.2f}%)")
        else:
            print(f"{status} {result.name}: {result.percent:.2f}% >= {result.minimum:.2f}% ({result.detail})")


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base", default="origin/main", help="Base ref for incremental coverage diff")
    parser.add_argument("--python-coverage", type=Path, default=Path(".coverage.json"))
    parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG)
    parser.add_argument("--skip-python", action="store_true")
    parser.add_argument("--skip-viewer-js", action="store_true")
    parser.add_argument("--skip-viewer-css", action="store_true")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    thresholds = load_thresholds(args.config)
    changed_lines = _filter_pure_viewer_asset_split(
        changed_lines_from_diff(_run_git_diff(args.base, VIEWER_DIFF_PATHS)),
        args.base,
    )

    results: list[CheckResult] = []
    if not args.skip_python:
        results.extend(
            check_python_coverage(
                args.python_coverage,
                changed_lines,
                thresholds["python_total_min"],
                thresholds["python_diff_min"],
            )
        )
    if not args.skip_viewer_js:
        results.extend(
            check_viewer_js_coverage(
                changed_viewer_functions(tuple(REPO_ROOT / source for source in VIEWER_JS_SOURCES), changed_lines),
                thresholds["viewer_js_function_min"],
                thresholds["viewer_js_diff_min"],
            )
        )
    if not args.skip_viewer_css:
        results.extend(
            check_viewer_css_coverage(
                changed_viewer_css_selectors(REPO_ROOT / VIEWER_CSS_SOURCE, changed_lines),
                thresholds["viewer_css_selector_min"],
                thresholds["viewer_css_diff_min"],
            )
        )

    print_results(results)
    return 0 if all(result.passed for result in results) else 2


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