vignette/scripts/check-dev-dashboard-ssot.py

#!/usr/bin/env python3
"""Validate that the dev dashboard stays aligned with current SSOT claims."""

from __future__ import annotations

import argparse
import json
import re
import shutil
import subprocess
from dataclasses import dataclass
from functools import lru_cache
from html.parser import HTMLParser
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_DASHBOARD = REPO_ROOT / "docs" / "dev_dashboard.html"
DEFAULT_SOURCE_GAPS = REPO_ROOT / "docs" / "guides" / "source-docs-and-gaps.md"
DEFAULT_BACKLOG = REPO_ROOT / "docs" / "ops" / "backlog-2026-06-26.md"
DEFAULT_TESTING = REPO_ROOT / "docs" / "guides" / "testing.md"
DEFAULT_LOCAL_DEVELOPMENT = REPO_ROOT / "docs" / "guides" / "local-development.md"
DEFAULT_TODO = REPO_ROOT / "docs" / "TODO.md"

EXPECTED_STATUS_COUNTS = {
    "done": 33,
    "doing": 2,
    "planned": 0,
}
EXPECTED_OWNER_COLUMN_COUNTS = {
    "block": 2,
    "decide": 0,
    "ext": 2,
}
EXPECTED_CRITICAL_CARD_COUNT = 3
EXPECTED_DECISION_DETAIL_ROWS = 0
EXPECTED_EXTERNAL_GATE_ROWS = 6
EXPECTED_OUTCOME_GATES = tuple(f"G{index}" for index in range(9))
EXPECTED_AOS_IDS = tuple(f"AOS-{index:03d}" for index in range(1, 13))
PLAYWRIGHT_ROOT = REPO_ROOT / "apps" / "web"

REQUIRED_SHARED_PHRASES = (
    "M2 digest worker + memory focused 30 passed",
    "M2 주변 회귀 87 passed",
    "one-shot worker/runner/default-off scheduler",
    "실 provider 장시간 운영",
    "임상 골든셋",
    "재압축",
    "블로커 2 · 결정 0 · 외부조율 2",
    "owner 결정 7건은 2026-06-30 전건 확정",
    "결정 0개(전건 확정)",
    "MASTERPLAN_REVISIONS F-10 승인",
    "Public API <b>119 paths · G1~G8</b>",
    "error code 1033",
    "Docker Desktop/DB 재기동",
    "watchdog healthy: engine, api, web-preview, cloudflared, public-api",
    "dev-dashboard E2E redteam 10 passed",
)

FORBIDDEN_STALE_PHRASES = (
    "M2 주변 회귀 82 passed",
    "one-shot digest worker + memory focused 25 passed",
    "M2 one-shot digest worker + memory focused 25 passed",
    "focused 26 passed",
    "운영 background scheduler",
    "최신 LTS",
    "최종 로컬 풀스택",
    "레이아웃 포커스(재설계 화면) | `session-layout`·`session-review`·`admin`·`learner`·`settings`·`teacher`, `@single-run` 제외 | desktop+mobile 병렬 | **58**",
    "M2는 보수적 identity/agreement pinned_fact 자동 실적재, append-only history, 명시적 상담 약속 철회 contradiction, episodic embedding writer, 다음 턴 EngineMessage 주입 회귀까지 완료했다.",
    "focused X2/backend 40 passed",
    "계획·블로커",
    "자유연습 기본값</b><p>수정안",
    "decisions 8건(owner 결정)",
    "결정 8개",
    "블로커 2 · 결정 7 · 외부조율 2",
    "decisions 7건(owner 결정)",
    "결정 7개",
    "블로커 2 · 결정 3 · 외부조율 2",
    "decisions 3건(owner 결정)",
    "결정 3개",
    "Public API <b>prod OK</b>",
    "Public API <b>530 BLOCK</b>",
    "API: prod/dev OK",
)


@dataclass(frozen=True)
class Check:
    name: str
    passed: bool
    detail: str


@dataclass(frozen=True)
class PlaywrightInventory:
    tests: int
    files: int


def _parse_playwright_inventory(output: str) -> PlaywrightInventory:
    match = re.search(r"^Total:\s+(\d+) tests in (\d+) files$", output, re.MULTILINE)
    if match is None:
        raise ValueError("playwright --list summary is missing")
    return PlaywrightInventory(tests=int(match.group(1)), files=int(match.group(2)))


@lru_cache(maxsize=1)
def _collect_playwright_inventory() -> PlaywrightInventory:
    executable = shutil.which("npx.cmd") or shutil.which("npx")
    if executable is None:
        raise RuntimeError("npx is unavailable for Playwright inventory")
    completed = subprocess.run(
        [executable, "playwright", "test", "--list"],
        cwd=PLAYWRIGHT_ROOT,
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        timeout=120,
    )
    if completed.returncode != 0:
        detail = (completed.stderr or completed.stdout).strip()
        raise RuntimeError(
            f"playwright --list failed ({completed.returncode}): {detail[-500:]}"
        )
    return _parse_playwright_inventory(completed.stdout)


class DashboardStructureParser(HTMLParser):
    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.status_counts: dict[str, int] = {
            status: 0 for status in EXPECTED_STATUS_COUNTS
        }
        self.owner_column_counts: dict[str, int] = {}
        self.decision_panel_rows = 0
        self.external_gate_rows = 0
        self._div_stack: list[str | None] = []
        self._owner_column_stack: list[str] = []
        self._decision_panel_stack: list[bool] = []

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        attrs_dict = dict(attrs)
        if tag == "div":
            owner_column = attrs_dict.get("data-owner-col")
            in_decision_panel = attrs_dict.get("id") == "panel-decisions" or (
                bool(self._decision_panel_stack) and self._decision_panel_stack[-1]
            )
            self._div_stack.append(owner_column)
            self._decision_panel_stack.append(in_decision_panel)
            if owner_column is not None:
                self._owner_column_stack.append(owner_column)
        if tag == "span":
            classes = attrs_dict.get("class", "")
            if "task-status" in classes and "s-plan" in classes:
                self.external_gate_rows += 1
            if (
                self._decision_panel_stack
                and self._decision_panel_stack[-1]
                and "task-status" in classes
                and "s-risk" in classes
            ):
                self.decision_panel_rows += 1
        if tag != "article":
            return

        status = attrs_dict.get("data-status")
        if status:
            self.status_counts[status] = self.status_counts.get(status, 0) + 1

        if attrs_dict.get("data-owner") != "1" or not self._owner_column_stack:
            return
        owner_column = self._owner_column_stack[-1]
        self.owner_column_counts[owner_column] = self.owner_column_counts.get(owner_column, 0) + 1

    def handle_endtag(self, tag: str) -> None:
        if tag != "div" or not self._div_stack:
            return
        owner_column = self._div_stack.pop()
        if self._decision_panel_stack:
            self._decision_panel_stack.pop()
        if owner_column is not None and self._owner_column_stack:
            self._owner_column_stack.pop()


def _read(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def _dashboard_structure(dashboard: str) -> DashboardStructureParser:
    parser = DashboardStructureParser()
    parser.feed(dashboard)
    parser.close()
    return parser


def _planned_cards(dashboard: str) -> list[str]:
    return re.findall(
        r'<article class="scard" data-status="planned"[\s\S]*?</article>',
        dashboard,
    )


def _has_check_status_markup(dashboard: str) -> bool:
    return bool(
        re.search(r'<[^>]+class="[^"]*\btask-status\b[^"]*\bs-check\b[^"]*"', dashboard)
        or re.search(r'>\s*CHECK\s*<', dashboard)
    )


def run_checks(
    paths: argparse.Namespace,
    *,
    playwright_inventory: PlaywrightInventory | None = None,
) -> dict[str, Any]:
    dashboard = _read(paths.dashboard)
    source_gaps = _read(paths.source_gaps)
    backlog = _read(paths.backlog)
    testing = _read(paths.testing)
    local_development = _read(paths.local_development)
    todo = _read(paths.todo)
    joined_docs = "\n".join([dashboard, source_gaps, backlog, testing, local_development])

    checks: list[Check] = []
    inventory_error: str | None = None
    if playwright_inventory is None:
        try:
            playwright_inventory = _collect_playwright_inventory()
        except (OSError, RuntimeError, ValueError, subprocess.SubprocessError) as exc:
            inventory_error = str(exc)
    if playwright_inventory is None:
        checks.append(
            Check(
                name="playwright_inventory_collected",
                passed=False,
                detail=inventory_error or "unknown inventory failure",
            )
        )
    else:
        inventory_phrase = (
            f"{playwright_inventory.tests} tests / {playwright_inventory.files} files"
        )
        checks.append(
            Check(
                name="playwright_inventory_collected",
                passed=True,
                detail=inventory_phrase,
            )
        )
        checks.append(
            Check(
                name="playwright_inventory_dashboard_current",
                passed=inventory_phrase in dashboard,
                detail=(
                    f"expected current inventory in dashboard: {inventory_phrase}"
                ),
            )
        )
        checks.append(
            Check(
                name="playwright_inventory_testing_guide_current",
                passed=inventory_phrase in testing,
                detail=(
                    f"expected current inventory in testing guide: {inventory_phrase}"
                ),
            )
        )
    stale_inventory_claims = (
        "현재 수집 기준 215 tests / 20 files",
        "**총 215 tests / 20 files**",
    )
    for phrase in stale_inventory_claims:
        checks.append(
            Check(
                name=f"stale_playwright_inventory::{phrase}",
                passed=phrase not in joined_docs,
                detail="absent" if phrase not in joined_docs else "present",
            )
        )
    structure = _dashboard_structure(dashboard)
    counts = structure.status_counts
    for status, expected in EXPECTED_STATUS_COUNTS.items():
        actual = counts.get(status, 0)
        checks.append(
            Check(
                name=f"status_count_{status}",
                passed=actual == expected,
                detail=f"expected={expected} actual={actual}",
            )
        )
    checks.append(
        Check(
            name="no_legacy_check_status_markup",
            passed=not _has_check_status_markup(dashboard),
            detail="task-status s-check/CHECK markup absent",
        )
    )

    planned = _planned_cards(dashboard)
    checks.append(
        Check(
            name="planned_cards_are_gate_framed",
            passed=all(("GATE" in card or "후속" in card or "남은" in card) for card in planned),
            detail=f"planned_cards={len(planned)}",
        )
    )

    outcome_gates = tuple(
        re.findall(
            r'<span class="chip c-(?:plan|doing|done)">(G[0-8]) (?:GATE|BUILD|DONE)(?: · [^<]+)?</span>',
            dashboard,
        )
    )
    checks.append(
        Check(
            name="outcome_alliance_gate_cards_complete",
            passed=outcome_gates == EXPECTED_OUTCOME_GATES,
            detail=f"expected={EXPECTED_OUTCOME_GATES} actual={outcome_gates}",
        )
    )
    todo_gate_headings = tuple(re.findall(r"^### I-[0-8]\. (G[0-8])\b", todo, re.MULTILINE))
    checks.append(
        Check(
            name="outcome_alliance_todo_gates_complete",
            passed=todo_gate_headings == EXPECTED_OUTCOME_GATES,
            detail=f"expected={EXPECTED_OUTCOME_GATES} actual={todo_gate_headings}",
        )
    )
    todo_aos_ids = tuple(re.findall(r"\*\*(AOS-\d{3})\*\*", todo))
    checks.append(
        Check(
            name="outcome_alliance_first_dag_complete",
            passed=todo_aos_ids == EXPECTED_AOS_IDS,
            detail=f"expected={EXPECTED_AOS_IDS} actual={todo_aos_ids}",
        )
    )

    for owner_column, expected in EXPECTED_OWNER_COLUMN_COUNTS.items():
        actual = structure.owner_column_counts.get(owner_column, 0)
        checks.append(
            Check(
                name=f"owner_column_count_{owner_column}",
                passed=actual == expected,
                detail=f"expected={expected} actual={actual}",
            )
        )
    expected_owner_total = sum(EXPECTED_OWNER_COLUMN_COUNTS.values())
    actual_owner_total = sum(structure.owner_column_counts.values())
    checks.append(
        Check(
            name="owner_action_total",
            passed=actual_owner_total == expected_owner_total,
            detail=f"expected={expected_owner_total} actual={actual_owner_total}",
        )
    )

    checks.append(
        Check(
            name="external_gate_rows_visible",
            passed=structure.external_gate_rows == EXPECTED_EXTERNAL_GATE_ROWS,
            detail=f"expected={EXPECTED_EXTERNAL_GATE_ROWS} actual={structure.external_gate_rows}",
        )
    )

    critical_card_count = dashboard.count('class="critbadge"')
    checks.append(
        Check(
            name="critical_card_count",
            passed=critical_card_count == EXPECTED_CRITICAL_CARD_COUNT,
            detail=f"expected={EXPECTED_CRITICAL_CARD_COUNT} actual={critical_card_count}",
        )
    )

    decision_detail_rows = structure.decision_panel_rows
    checks.append(
        Check(
            name="decision_detail_rows",
            passed=decision_detail_rows == EXPECTED_DECISION_DETAIL_ROWS,
            detail=f"expected={EXPECTED_DECISION_DETAIL_ROWS} actual={decision_detail_rows}",
        )
    )

    for phrase in REQUIRED_SHARED_PHRASES:
        checks.append(
            Check(
                name=f"required_phrase::{phrase}",
                passed=phrase in joined_docs,
                detail="present" if phrase in joined_docs else "missing",
            )
        )
    for phrase in FORBIDDEN_STALE_PHRASES:
        checks.append(
            Check(
                name=f"forbidden_phrase::{phrase}",
                passed=phrase not in joined_docs,
                detail="absent" if phrase not in joined_docs else "present",
            )
        )

    return {
        "schema": "vignette.dev_dashboard_ssot_check.v1",
        "passed": all(check.passed for check in checks),
        "status_counts": counts,
        "checks": [check.__dict__ for check in checks],
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--dashboard", type=Path, default=DEFAULT_DASHBOARD)
    parser.add_argument("--source-gaps", type=Path, default=DEFAULT_SOURCE_GAPS)
    parser.add_argument("--backlog", type=Path, default=DEFAULT_BACKLOG)
    parser.add_argument("--testing", type=Path, default=DEFAULT_TESTING)
    parser.add_argument("--local-development", type=Path, default=DEFAULT_LOCAL_DEVELOPMENT)
    parser.add_argument("--todo", type=Path, default=DEFAULT_TODO)
    parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON.")
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    report = run_checks(args)
    if args.json:
        print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
    else:
        status = "PASS" if report["passed"] else "FAIL"
        print(f"{status} dev dashboard SSOT check")
        for check in report["checks"]:
            marker = "PASS" if check["passed"] else "FAIL"
            print(f"{marker} {check['name']} {check['detail']}")
    return 0 if report["passed"] else 1


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