#!/usr/bin/env bash

set -euo pipefail

ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
source "$ROOT_DIR/scripts/harness/lib.sh"

main() {
  cd "$ROOT_DIR"
  local uv_bin
  uv_bin="$(harness_require_tool uv)"

  harness_run "$uv_bin" run --frozen --project "$ROOT_DIR" python - <<'PY'
import ast
import re
from pathlib import Path
from urllib.parse import parse_qs, urlsplit

root = Path.cwd()
failures: list[str] = []
warnings: list[str] = []


def fail(message: str) -> None:
    failures.append(message)


def warn(message: str) -> None:
    warnings.append(message)


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


def literal_assignment(path: str, name: str) -> object:
    tree = ast.parse(read_text(path))
    for node in tree.body:
        if not isinstance(node, ast.Assign):
            continue
        if not any(isinstance(target, ast.Name) and target.id == name for target in node.targets):
            continue
        return ast.literal_eval(node.value)
    raise ValueError(f"could not find literal assignment {name} in {path}")


def slug_from_filename(filename: str) -> str:
    return "overview" if filename == "README.md" else filename.removesuffix(".md")


def frontend_docs_metadata() -> dict[str, dict[str, object]]:
    text = read_text("frontend/src/pages/Docs/docs-content.ts")
    entries: dict[str, dict[str, object]] = {}
    pattern = re.compile(
        r"(?:\[(?P<constant>DOCS_README_FILENAME)\]|'(?P<filename>[^']+\.md)'):\s*\{\s*"
        r"zh:\s*\{\s*title:\s*'(?P<zh_title>[^']+)',\s*group:\s*'(?P<zh_group>[^']+)',\s*order:\s*(?P<zh_order>\d+)\s*\},\s*"
        r"en:\s*\{\s*title:\s*'(?P<en_title>[^']+)',\s*group:\s*'(?P<en_group>[^']+)',\s*order:\s*(?P<en_order>\d+)\s*\}",
        re.S,
    )
    for match in pattern.finditer(text):
        filename = "README.md" if match.group("constant") else match.group("filename")
        if filename in entries:
            fail(f"frontend Docs metadata registers {filename} more than once")
            continue
        entries[filename] = {
            "slug": slug_from_filename(filename),
            "zh_title": match.group("zh_title"),
            "en_title": match.group("en_title"),
            "zh_group": match.group("zh_group"),
            "en_group": match.group("en_group"),
            "zh_order": int(match.group("zh_order")),
            "en_order": int(match.group("en_order")),
        }
    return entries


def backend_docs_metadata() -> dict[str, dict[str, object]]:
    tree = ast.parse(read_text("backend/app/services/docs_gatekeeper.py"))
    constants: dict[str, object] = {}

    for node in tree.body:
        if not isinstance(node, ast.Assign) or len(node.targets) != 1:
            continue
        target = node.targets[0]
        if isinstance(target, ast.Name) and isinstance(node.value, ast.Constant):
            constants[target.id] = node.value.value

    def read_arg(node: ast.AST) -> object:
        if isinstance(node, ast.Constant):
            return node.value
        if isinstance(node, ast.Name) and node.id in constants:
            return constants[node.id]
        raise ValueError(f"unsupported DocsMetadata argument: {ast.dump(node)}")

    entries: dict[str, dict[str, object]] = {}
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        if not isinstance(node.func, ast.Name) or node.func.id != "DocsMetadata":
            continue
        if len(node.args) != 7:
            fail("backend DocsMetadata entries must keep the expected seven positional fields")
            continue
        filename, slug, access, group, order, zh_title, en_title = [read_arg(arg) for arg in node.args]
        if not isinstance(filename, str):
            fail("backend DocsMetadata filename must be a string")
            continue
        if filename in entries:
            fail(f"backend Docs metadata registers {filename} more than once")
            continue
        entries[filename] = {
            "slug": slug,
            "access": access,
            "group": group,
            "order": order,
            "zh_title": zh_title,
            "en_title": en_title,
        }
    return entries


def docs_metadata_filenames() -> set[str]:
    return set(frontend_docs_metadata())


def check_docs_metadata_alignment() -> None:
    frontend = frontend_docs_metadata()
    backend = backend_docs_metadata()

    if not frontend:
        fail("frontend Docs metadata must expose at least one document")
    if not backend:
        fail("backend Gatekeeper Docs metadata must expose at least one document")

    frontend_only = sorted(set(frontend) - set(backend))
    backend_only = sorted(set(backend) - set(frontend))
    if frontend_only:
        fail("frontend Docs metadata has files missing from backend Gatekeeper metadata: " + ", ".join(frontend_only))
    if backend_only:
        fail("backend Gatekeeper metadata has files missing from frontend Docs metadata: " + ", ".join(backend_only))

    for filename in sorted(set(frontend) & set(backend)):
        frontend_entry = frontend[filename]
        backend_entry = backend[filename]
        if frontend_entry["zh_group"] != frontend_entry["en_group"]:
            fail(f"frontend Docs metadata group differs by language for {filename}")
        if frontend_entry["zh_order"] != frontend_entry["en_order"]:
            fail(f"frontend Docs metadata order differs by language for {filename}")

        expected = {
            "slug": frontend_entry["slug"],
            "group": frontend_entry["zh_group"],
            "order": frontend_entry["zh_order"],
            "zh_title": frontend_entry["zh_title"],
            "en_title": frontend_entry["en_title"],
        }
        actual = {key: backend_entry[key] for key in expected}
        if actual != expected:
            fail(f"{filename}: frontend Docs metadata and backend Gatekeeper metadata differ; frontend={expected}, backend={actual}")


def check_public_docs_registry() -> None:
    filenames = docs_metadata_filenames()
    for filename in sorted(filenames):
        for lang in ("zh", "en"):
            path = root / "docs/technical" / lang / filename
            if not path.exists():
                fail(f"public docs metadata references missing file: {path.relative_to(root)}")

    for readme in (root / "docs/technical/zh/README.md", root / "docs/technical/en/README.md"):
        if not readme.exists():
            continue
        for href in re.findall(r"\]\(([^)]+\.md)\)", readme.read_text(encoding="utf-8")):
            if "docs/technical/" not in href:
                continue
            filename = Path(href).name
            if filename not in filenames:
                fail(f"{readme.relative_to(root)} links {filename}, but DOCS_METADATA does not expose it")


def check_all_technical_docs_have_bilingual_pairs() -> None:
    zh_dir = root / "docs/technical/zh"
    en_dir = root / "docs/technical/en"
    zh_files = {path.name for path in zh_dir.glob("*.md")}
    en_files = {path.name for path in en_dir.glob("*.md")}

    zh_only = sorted(zh_files - en_files)
    en_only = sorted(en_files - zh_files)
    if zh_only:
        fail("technical docs missing English counterparts: " + ", ".join(zh_only))
    if en_only:
        fail("technical docs missing Chinese counterparts: " + ", ".join(en_only))


def check_bilingual_docs_are_not_copies() -> None:
    same: list[str] = []
    for en in sorted((root / "docs/technical/en").glob("*.md")):
        zh = root / "docs/technical/zh" / en.name
        if zh.exists() and en.read_text(encoding="utf-8") == zh.read_text(encoding="utf-8"):
            same.append(en.name)
    if same:
        fail("identical en/zh technical docs: " + ", ".join(same))


def check_public_doc_links_exist() -> None:
    repo_prefix = f"{root}/"
    for doc in sorted((root / "docs/technical").glob("*/*.md")):
        text = doc.read_text(encoding="utf-8")
        for href in re.findall(r"\]\(([^)#]+\.md)(?:#[^)]+)?\)", text):
            if not href.startswith(repo_prefix):
                continue
            target = Path(href)
            if not target.exists():
                fail(f"{doc.relative_to(root)} links missing markdown file: {href}")


def check_language_scoped_technical_links() -> None:
    technical_docs_prefix = re.escape(str(root / "docs/technical"))
    pattern = re.compile(rf"{technical_docs_prefix}/(?!zh/|en/)[^)#\s]+")
    for doc in sorted((root / "docs/technical").glob("*/*.md")):
        text = doc.read_text(encoding="utf-8")
        for match in pattern.finditer(text):
            line_no = text.count("\n", 0, match.start()) + 1
            fail(
                f"{doc.relative_to(root)}:{line_no}: technical doc link must include "
                f"the language directory: {match.group(0)}"
            )


def check_public_doc_link_titles() -> None:
    pattern = re.compile(r"\[([^]\n]+\.md)\]\(")
    for doc in sorted((root / "docs/technical").glob("*/*.md")):
        text = doc.read_text(encoding="utf-8")
        for match in pattern.finditer(text):
            line_no = text.count("\n", 0, match.start()) + 1
            fail(
                f"{doc.relative_to(root)}:{line_no}: public docs should use readable "
                f"link text instead of raw filename {match.group(1)!r}"
            )


def check_credential_collector_contracts() -> None:
    defaults = literal_assignment("backend/app/core/datasource_defaults.py", "DEFAULT_DATASOURCES")
    if not isinstance(defaults, dict):
        fail("backend/app/core/datasource_defaults.py DEFAULT_DATASOURCES must stay a dict")
        return

    supported_sources: dict[str, str] = {}
    for source, info in defaults.items():
        if not isinstance(info, dict):
            fail(f"DEFAULT_DATASOURCES entry {source!r} must be a dict")
            continue
        if not info.get("requires_credentials"):
            continue
        if info.get("credential_status") != "supported":
            continue
        provider = info.get("credential_provider")
        if not isinstance(provider, str) or not provider.strip():
            fail(f"{source}: supported credential collector is missing credential_provider")
            continue
        supported_sources[str(source)] = provider

    if not supported_sources:
        return

    guides_text = read_text("backend/app/services/credential_guides.py")
    default_guides = set(re.findall(r"CredentialGuideDefault\(\s*provider=\"([^\"]+)\"", guides_text))
    connectivity_providers = literal_assignment(
        "backend/app/services/datasource_connectivity.py",
        "SUPPORTED_CREDENTIAL_PROVIDERS",
    )
    if not isinstance(connectivity_providers, set):
        fail("SUPPORTED_CREDENTIAL_PROVIDERS must stay a literal set")
        connectivity_providers = set()

    frontend_text = read_text("frontend/src/admin/pages/PlainResourcePages.tsx")
    tests_text = read_text("backend/tests/test_collectors.py")
    zh_doc = read_text("docs/technical/zh/datasource-collector-settings-connectivity.md")
    en_doc = read_text("docs/technical/en/datasource-collector-settings-connectivity.md")

    if "test_supported_credential_collectors_have_guides_and_connectivity_provider" not in tests_text:
        fail("backend/tests/test_collectors.py must keep the supported credential collector contract test")

    if "loadCredentialGuide" not in frontend_text or "credentialGuideProvider" not in frontend_text:
        fail("collector credential UI must keep guide-loading and provider-normalization helpers")

    for source, provider in sorted(supported_sources.items()):
        if provider not in default_guides:
            fail(f"{source}: missing default credential guide for provider {provider}")
        if provider not in connectivity_providers:
            fail(f"{source}: missing supported connectivity provider {provider}")
        if provider not in frontend_text and source not in frontend_text:
            fail(f"{source}: collector credential UI does not mention provider/source {provider}")
        for doc_path, doc_text in (
            ("docs/technical/zh/datasource-collector-settings-connectivity.md", zh_doc),
            ("docs/technical/en/datasource-collector-settings-connectivity.md", en_doc),
        ):
            if provider not in doc_text and source not in doc_text:
                fail(f"{doc_path}: missing supported credential collector provider/source {provider}/{source}")


def known_frontend_routes() -> set[str]:
    app_text = read_text("frontend/src/App.tsx")
    admin_text = read_text("frontend/src/admin/AdminRoutes.tsx")
    manifest_text = read_text("frontend/src/admin/routes/manifest.tsx")
    routes = set(re.findall(r'<Route\s+path="([^"*][^"]*)"', app_text))
    routes.update(re.findall(r'<Route\s+path="([^"*][^"]*)"', admin_text))
    routes.update(re.findall(r"path:\s*'([^']+)'", manifest_text))
    routes.update({"/", "/playground", "/docs/:slug"})
    routes.discard("*")
    return routes


def admin_manifest_routes() -> dict[str, str]:
    manifest_text = read_text("frontend/src/admin/routes/manifest.tsx")
    routes: dict[str, str] = {}
    pattern = re.compile(r"\{\s*path:\s*'([^']+)',\s*label:\s*'([^']+)'", re.S)
    for path, label in pattern.findall(manifest_text):
        routes[path] = label
    if not routes:
        fail("could not parse admin route manifest for manual coverage checks")
    return routes


def manual_console_route_rows(doc_path: Path, heading: str) -> dict[str, str]:
    text = doc_path.read_text(encoding="utf-8")
    heading_index = text.find(heading)
    if heading_index == -1:
        fail(f"{doc_path.relative_to(root)} is missing {heading!r}")
        return {}
    next_heading = re.search(r"\n##\s+", text[heading_index + len(heading):])
    section = text[heading_index:] if not next_heading else text[heading_index:heading_index + len(heading) + next_heading.start()]
    rows: dict[str, str] = {}
    for line in section.splitlines():
        match = re.match(r"\|\s*([^|`][^|]*?)\s*\|\s*`([^`]+)`\s*\|", line)
        if not match:
            continue
        label = match.group(1).strip()
        path = match.group(2).strip()
        rows[path] = label
    return rows


def normalize_zh_label(label: str) -> str:
    return re.sub(r"\s+", "", label)


def check_manual_console_route_tables() -> None:
    manifest = admin_manifest_routes()
    zh_manual = root / "docs/technical/zh/manual.md"
    en_manual = root / "docs/technical/en/manual.md"
    manual_rows = {
        "zh": manual_console_route_rows(zh_manual, "## 控制台总览"),
        "en": manual_console_route_rows(en_manual, "## Console Overview"),
    }

    for lang, rows in manual_rows.items():
        missing = sorted(set(manifest) - set(rows))
        extra = sorted(set(rows) - set(manifest))
        if missing:
            fail(f"docs/technical/{lang}/manual.md console overview misses admin manifest route(s): " + ", ".join(missing))
        if extra:
            fail(f"docs/technical/{lang}/manual.md console overview lists route(s) missing from admin manifest: " + ", ".join(extra))

    zh_rows = manual_rows["zh"]
    for path, expected_label in manifest.items():
        actual = zh_rows.get(path)
        if actual is None:
            continue
        if normalize_zh_label(actual) != normalize_zh_label(expected_label):
            fail(
                "docs/technical/zh/manual.md console overview label mismatch for "
                f"{path}: expected {expected_label!r}, got {actual!r}"
            )


def extract_balanced(text: str, start: int, open_char: str, close_char: str) -> str:
    depth = 0
    quote: str | None = None
    escape = False
    for index in range(start, len(text)):
        char = text[index]
        if quote:
            if escape:
                escape = False
            elif char == "\\":
                escape = True
            elif char == quote:
                quote = None
            continue
        if char in ("'", '"', "`"):
            quote = char
            continue
        if char == open_char:
            depth += 1
        elif char == close_char:
            depth -= 1
            if depth == 0:
                return text[start:index + 1]
    raise ValueError(f"could not find balanced {open_char}{close_char} block")


def find_balanced_after(text: str, marker: str, open_char: str, close_char: str) -> str:
    marker_index = text.index(marker)
    start = text.index(open_char, marker_index)
    return extract_balanced(text, start, open_char, close_char)


def nesting_depth(text: str, stop: int, open_char: str, close_char: str) -> int:
    depth = 0
    quote: str | None = None
    escape = False
    for char in text[:stop]:
        if quote:
            if escape:
                escape = False
            elif char == "\\":
                escape = True
            elif char == quote:
                quote = None
            continue
        if char in ("'", '"', "`"):
            quote = char
            continue
        if char == open_char:
            depth += 1
        elif char == close_char:
            depth -= 1
    return depth


def top_level_config_blocks(configs_block: str) -> dict[str, str]:
    blocks: dict[str, str] = {}
    index = 1
    while index < len(configs_block) - 1:
        match = re.search(r"\b([A-Za-z][A-Za-z0-9_]*)\s*:\s*\{", configs_block[index:])
        if not match:
            break
        name = match.group(1)
        start = index + match.end() - 1
        if nesting_depth(configs_block, start, "{", "}") != 1:
            index = start + 1
            continue
        block = extract_balanced(configs_block, start, "{", "}")
        blocks[name] = block
        index = start + len(block)
    return blocks


def direct_section_keys(config_block: str) -> set[str]:
    sections_match = re.search(r"\bsections\s*:\s*\[", config_block)
    if not sections_match:
        return set()
    sections_block = extract_balanced(config_block, sections_match.end() - 1, "[", "]")
    keys: set[str] = set()
    index = 1
    while index < len(sections_block) - 1:
        match = re.search(r"\{\s*key\s*:\s*'([^']+)'", sections_block[index:])
        if not match:
            break
        start = index + match.start()
        if (
            nesting_depth(sections_block, start, "[", "]") == 1
            and nesting_depth(sections_block, start, "{", "}") == 0
        ):
            keys.add(match.group(1))
            block = extract_balanced(sections_block, start, "{", "}")
            index = start + len(block)
        else:
            index = start + 1
    return keys


def known_admin_sections() -> dict[str, set[str]]:
    plain_text = read_text("frontend/src/admin/pages/PlainResourcePages.tsx")
    admin_text = read_text("frontend/src/admin/AdminRoutes.tsx")

    configs_block = find_balanced_after(plain_text, "const configs =", "{", "}")
    configs = top_level_config_blocks(configs_block)
    component_to_config = dict(re.findall(
        r"export function (\w+)\(\) \{\s*return <ModuleConsole config=\{configs\.([A-Za-z0-9_]+)\} />\s*\}",
        plain_text,
    ))
    path_to_component = dict(re.findall(
        r'<Route\s+path="([^"]+)"\s+element=\{<([A-Za-z0-9_]+)\s*/>\}',
        admin_text,
    ))

    section_map: dict[str, set[str]] = {}
    for path, component in path_to_component.items():
        config_key = component_to_config.get(component)
        if not config_key:
            continue
        if config_key not in configs:
            fail(f"Admin route {path} uses unknown PlainResourcePages config: {config_key}")
            continue
        section_map[path] = direct_section_keys(configs[config_key])
    return section_map


def check_documented_section_deep_links() -> None:
    section_map = known_admin_sections()
    pattern = re.compile(r"/[a-z][a-z0-9/-]*\?section=[a-z0-9_/-]+")
    docs = [
        *sorted((root / "docs/technical").glob("*/*.md")),
        *sorted((root / "docs/plans").glob("*.md")),
    ]
    for doc in docs:
        text = doc.read_text(encoding="utf-8")
        for match in pattern.finditer(text):
            value = match.group(0)
            parsed = urlsplit(value)
            section = parse_qs(parsed.query).get("section", [""])[0]
            if parsed.path not in section_map:
                line_no = text.count("\n", 0, match.start()) + 1
                fail(f"{doc.relative_to(root)}:{line_no}: documented section link uses a route with no known sections: {value}")
                continue
            if section not in section_map[parsed.path]:
                line_no = text.count("\n", 0, match.start()) + 1
                known = ", ".join(sorted(section_map[parsed.path]))
                fail(f"{doc.relative_to(root)}:{line_no}: documented section link {value} is not a known section; expected one of: {known}")


def check_documented_ui_routes() -> None:
    routes = known_frontend_routes()
    allowed_prefixes = (
        "/api/",
        "/ws",
        "/health",
        "/legacy/",
        "/earth/",
        "/docs/",
        "/assets/",
        "/components/",
        "/dev/",
        "/etc/",
        "/home/",
        "/tmp/",
        "/root",
        "/app/",
        "/planet",
        "/localhost",
        "/example",
        "/your-",
    )
    route_docs = [
        root / "docs/technical/zh/README.md",
        root / "docs/technical/en/README.md",
        root / "docs/technical/zh/manual.md",
        root / "docs/technical/en/manual.md",
        root / "docs/technical/zh/quickstart.md",
        root / "docs/technical/en/quickstart.md",
        root / "docs/technical/zh/frontend-admin-frontend-context.md",
        root / "docs/technical/en/frontend-admin-frontend-context.md",
    ]
    route_pattern = re.compile(r"(?<![\w:_}-])/(?:[a-z][a-z0-9-]*)(?:/[a-z0-9:_-]+)*")
    for doc in route_docs:
        if not doc.exists():
            continue
        text = doc.read_text(encoding="utf-8")
        for candidate in sorted(set(route_pattern.findall(text))):
            if candidate in routes:
                continue
            if candidate.startswith(allowed_prefixes):
                continue
            if re.search(r"/(start|stop|restart|reset|status|stream|generate|cancel|tasks?|task-status|events)$", candidate):
                continue
            if re.fullmatch(r"/[a-z]{2,3}", candidate):
                continue
            if candidate == "/admin-next" or candidate.startswith("/admin-next/"):
                fail(f"{doc.relative_to(root)} references stale admin-next route {candidate}")
            elif candidate.startswith("/") and candidate.count("/") <= 2:
                warn(f"{doc.relative_to(root)} references route-like path not in current frontend routes: {candidate}")


def check_stale_admin_doc_terms() -> None:
    stale_patterns = [
        (re.compile(r"React\s*\+\s*Ant Design"), "console frontend is no longer React + Ant Design"),
        (re.compile(r"Ant Design Pro"), "console frontend is no longer Ant Design Pro"),
        (re.compile(r"Old AntD legacy pages"), "old AntD legacy pages are not part of the active frontend"),
        (re.compile(r"Task Queue:\s*Celery"), "Celery is not part of the active local task stack"),
        (re.compile(r"Message Queue:\s*Kafka"), "Kafka is not part of the active local message stack"),
        (re.compile(r"UI Library:\s*Ant Design Pro"), "console frontend is no longer Ant Design Pro"),
        (re.compile(r"├── unreal/"), "UE5 is not an active checked-in local project tree"),
        (re.compile(r"Unreal Engine 5 3D visualization"), "UE5 is future/optional, not the active local visualization shell"),
        (re.compile(r"Polarized 3D large display \(4K, 120Hz\)"), "physical display work is future/optional, not the active local loop"),
        (re.compile(r"\?tab="), "admin deep links use ?section=, not ?tab="),
    ]
    docs = [
        root / "AGENTS.md",
        root / "README.md",
        root / "project_context.md",
        *sorted((root / "docs/technical").glob("*/*.md")),
        *sorted((root / "docs/plans").glob("*.md")),
    ]
    for doc in docs:
        text = doc.read_text(encoding="utf-8")
        for pattern, reason in stale_patterns:
            for match in pattern.finditer(text):
                line_no = text.count("\n", 0, match.start()) + 1
                fail(f"{doc.relative_to(root)}:{line_no}: stale admin documentation: {reason}")


def check_harness_rule_coverage_notes() -> None:
    harness_doc = read_text("docs/HARNESS.md")
    audit_doc = read_text("docs/harness-audit.md")
    if "Rules Coverage Evidence" not in harness_doc:
        fail("docs/HARNESS.md must point frontend/docs audits to the Rules Coverage Evidence section")
    if "## Rules Coverage Evidence" not in audit_doc:
        fail("docs/harness-audit.md must keep the Rules Coverage Evidence section")

    for module in ("core", "security", "workflow", "docs", "uiux", "frontend", "earth"):
        if f"| `{module}` |" not in audit_doc:
            fail(f"docs/harness-audit.md Rules Coverage Evidence must include the `{module}` rules module")


def main() -> None:
    check_docs_metadata_alignment()
    check_public_docs_registry()
    check_all_technical_docs_have_bilingual_pairs()
    check_bilingual_docs_are_not_copies()
    check_public_doc_links_exist()
    check_language_scoped_technical_links()
    check_public_doc_link_titles()
    check_credential_collector_contracts()
    check_manual_console_route_tables()
    check_documented_section_deep_links()
    check_documented_ui_routes()
    check_stale_admin_doc_terms()
    check_harness_rule_coverage_notes()

    for message in warnings:
        print(f"warn: {message}")
    if failures:
        for message in failures:
            print(f"fail: {message}")
        raise SystemExit(f"docs consistency check failed: {len(failures)} failure(s), {len(warnings)} warning(s)")
    print(f"docs consistency check passed: {len(warnings)} warning(s)")


main()
PY
}

main "$@"