#!/usr/bin/env python3
"""Read-only architecture inventory for AirShelf.

The scanner intentionally has no apply/write mode. It emits deterministic JSON to
stdout and limits all content reads to architecture-relevant, non-sensitive files.
"""

from __future__ import annotations

import argparse
import ast
import base64
import fnmatch
import hashlib
import json
import os
import re
import subprocess
import sys
import zlib
from pathlib import Path, PurePosixPath
from typing import Any, Callable, Iterable, Sequence
from urllib.parse import unquote


SCHEMA_VERSION = 1
MAX_TEXT_BYTES = 1_000_000
MAX_HASH_BYTES = 10_000_000
TARGET_DOCUMENTS = (
    "core/backend/ARCHITECTURE.md",
    "core/frontend/ARCHITECTURE.md",
)
GIT_SCOPES = ("core/backend", "core/frontend", "k8s/core")
RELEVANT_PATTERNS = (
    "core/backend/airshelf/settings/*.py",
    "core/backend/airshelf/urls.py",
    "core/backend/apps/*/apps.py",
    "core/backend/apps/*/models.py",
    "core/backend/apps/*/urls.py",
    "core/backend/apps/*/tasks.py",
    "core/backend/apps/ai/providers/*.py",
    "core/backend/requirements.txt",
    "core/backend/Dockerfile",
    "core/backend/docker-entrypoint.sh",
    "k8s/core/*",
    "k8s/core/**/*",
    "core/frontend/package.json",
    "core/frontend/src/App.tsx",
    "core/frontend/src/api.ts",
    "core/frontend/src/types.ts",
    "core/frontend/src/routes/*",
    "core/frontend/src/routes/**/*",
    "core/frontend/src/components/*",
    "core/frontend/src/components/**/*",
    "core/frontend/vite.config.ts",
    "core/frontend/Dockerfile",
    "core/frontend/nginx.conf",
)
EXCLUDED_PARTS = frozenset(
    {
        ".git",
        ".venv",
        "venv",
        "node_modules",
        "dist",
        "build",
        ".next",
        "coverage",
        "__pycache__",
        ".pytest_cache",
        ".mypy_cache",
        ".ruff_cache",
        ".cache",
        "cache",
        "media",
        "uploads",
        "user_uploads",
        "staticfiles",
        "ab_results",
    }
)
SENSITIVE_NAMES = frozenset(
    {
        "account.md",
        "credentials.json",
        "service-account.json",
        "secrets.json",
        "id_rsa",
        "id_ed25519",
    }
)
SENSITIVE_SUFFIXES = (
    ".pem",
    ".key",
    ".p12",
    ".pfx",
    ".sqlite",
    ".sqlite3",
    ".db",
)
SYNC_MARKER_RE = re.compile(
    r"<!--\s*architecture-sync-commit:\s*([0-9a-fA-F]{7,40})\s*-->"
)
HEADING_RE = re.compile(r"^(#{1,6})\s+(.+?)\s*$", re.MULTILINE)
TS_EXPORT_RE = re.compile(
    r"\bexport\s+(?:default\s+)?(?:async\s+)?"
    r"(?:const|let|var|function|class|type|interface|enum)\s+([A-Za-z_$][\w$]*)"
)
QUOTED_PATH_RE = re.compile(r"[\"'`](/[^\"'`\s]*)[\"'`]")
MARKDOWN_LINK_RE = re.compile(r"!?\[[^\]]*\]\(([^)]+)\)")
MARKDOWN_REFERENCE_LINK_RE = re.compile(r"(?m)^\s*\[[^\]]+\]:\s*(\S+)")
ABSOLUTE_LOCAL_PATH_RE = re.compile(
    r"(?i)(?:\b[A-Z]:[\\/]|file://|/(?:Users|home|var/folders|private/tmp)/)"
)
SECRET_CONTENT_PATTERNS = (
    re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
    re.compile(r"\bsk-[A-Za-z0-9_-]{20,}\b"),
    re.compile(
        r"(?i)\b(?:api[_-]?key|secret|password)\b\s*[:=]\s*[\"']?[A-Za-z0-9_./+=-]{12,}"
    ),
)
BACKEND_DOCUMENT = "core/backend/ARCHITECTURE.md"
FRONTEND_DOCUMENT = "core/frontend/ARCHITECTURE.md"
BACKEND_APP_SECTIONS = {
    "common": "4.1 `common`",
    "accounts": "4.2 `accounts`",
    "products": "4.3 `products`",
    "assets": "4.4 `assets`",
    "projects": "4.5 `projects`",
    "ai": "4.6 `ai`",
    "billing": "4.7 `billing`",
    "ops": "4.8 `ops`",
    "adminpanel": "4.9 `adminpanel`",
}


class ScanError(RuntimeError):
    """A safe, user-reportable scanner failure."""


def normalize_path(value: str) -> str:
    path = value.replace("\\", "/").removeprefix("./")
    pure = PurePosixPath(path)
    if not path or pure.is_absolute() or ".." in pure.parts:
        return ""
    return pure.as_posix()


def is_forbidden(path: str) -> bool:
    normalized = normalize_path(path)
    if not normalized:
        return True
    parts = tuple(part.lower() for part in PurePosixPath(normalized).parts)
    name = parts[-1]
    if any(part in EXCLUDED_PARTS for part in parts):
        return True
    if name == ".env" or name.startswith(".env."):
        return True
    if name in SENSITIVE_NAMES or name.endswith(SENSITIVE_SUFFIXES):
        return True
    return False


def is_relevant(path: str) -> bool:
    normalized = normalize_path(path)
    return bool(normalized) and not is_forbidden(normalized) and any(
        fnmatch.fnmatchcase(normalized, pattern) for pattern in RELEVANT_PATTERNS
    )


def selected_documents(scope: str) -> tuple[str, ...]:
    if scope == "backend":
        return (BACKEND_DOCUMENT,)
    if scope == "frontend":
        return (FRONTEND_DOCUMENT,)
    return TARGET_DOCUMENTS


def path_in_scope(path: str, scope: str) -> bool:
    if scope == "backend":
        return path.startswith("core/backend/") or path.startswith("k8s/core/")
    if scope == "frontend":
        return path.startswith("core/frontend/")
    return path.startswith(("core/backend/", "core/frontend/", "k8s/core/"))


def filter_entries(entries: Iterable[dict[str, str]], scope: str) -> list[dict[str, str]]:
    return [entry for entry in entries if path_in_scope(entry["path"], scope)]


def safe_candidate(root: Path, relative: str) -> Path | None:
    normalized = normalize_path(relative)
    if not normalized or is_forbidden(normalized):
        return None
    candidate = root.joinpath(*PurePosixPath(normalized).parts)
    try:
        resolved_root = root.resolve(strict=True)
        resolved = candidate.resolve(strict=False)
        resolved.relative_to(resolved_root)
    except (OSError, ValueError):
        return None
    current = candidate
    while current != root:
        if current.is_symlink():
            return None
        current = current.parent
    return candidate


def read_bytes(root: Path, relative: str, warnings: set[str]) -> bytes | None:
    candidate = safe_candidate(root, relative)
    if candidate is None or not candidate.is_file():
        return None
    try:
        size = candidate.stat().st_size
        if size > MAX_TEXT_BYTES:
            warnings.add(f"file_too_large:{normalize_path(relative)}")
            return None
        return candidate.read_bytes()
    except OSError:
        warnings.add(f"file_unreadable:{normalize_path(relative)}")
        return None


def read_text(root: Path, relative: str, warnings: set[str]) -> str | None:
    data = read_bytes(root, relative, warnings)
    if data is None:
        return None
    try:
        return data.decode("utf-8-sig")
    except UnicodeDecodeError:
        warnings.add(f"file_not_utf8:{normalize_path(relative)}")
        return None


def file_fingerprint(root: Path, relative: str, warnings: set[str]) -> str | None:
    candidate = safe_candidate(root, relative)
    if candidate is None or not candidate.is_file():
        return None
    try:
        if candidate.stat().st_size > MAX_HASH_BYTES:
            warnings.add(f"fingerprint_file_too_large:{normalize_path(relative)}")
            return None
        digest = hashlib.sha256()
        with candidate.open("rb") as stream:
            for chunk in iter(lambda: stream.read(131_072), b""):
                digest.update(chunk)
        return digest.hexdigest()
    except OSError:
        warnings.add(f"fingerprint_unavailable:{normalize_path(relative)}")
        return None


def git(root: Path, arguments: Sequence[str], *, required: bool = True) -> bytes:
    environment = os.environ.copy()
    environment["GIT_OPTIONAL_LOCKS"] = "0"
    try:
        result = subprocess.run(
            ["git", "-c", "core.quotepath=false", *arguments],
            cwd=root,
            env=environment,
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=False,
        )
    except OSError as error:
        raise ScanError("git_unavailable") from error
    if result.returncode and required:
        raise ScanError(f"git_command_failed:{arguments[0]}")
    return result.stdout if result.returncode == 0 else b""


def locate_repository(start: Path) -> Path:
    output = git(start, ("rev-parse", "--show-toplevel"))
    try:
        root = Path(output.decode("utf-8").strip()).resolve(strict=True)
    except (UnicodeDecodeError, OSError) as error:
        raise ScanError("repository_root_unavailable") from error
    if not root.is_dir():
        raise ScanError("repository_root_unavailable")
    return root


def decode_git_token(token: bytes) -> str:
    return token.decode("utf-8", errors="replace")


def parse_name_status(
    data: bytes, path_filter: Callable[[str], bool] = is_relevant
) -> list[dict[str, str]]:
    tokens = [decode_git_token(token) for token in data.split(b"\0") if token]
    entries: list[dict[str, str]] = []
    index = 0
    while index < len(tokens):
        status = tokens[index]
        index += 1
        if index >= len(tokens):
            break
        if status.startswith(("R", "C")):
            old_path = normalize_path(tokens[index])
            index += 1
            if index >= len(tokens):
                break
            new_path = normalize_path(tokens[index])
            index += 1
            if path_filter(new_path):
                entry = {"path": new_path, "status": status}
                if path_filter(old_path):
                    entry["old_path"] = old_path
                entries.append(entry)
            elif path_filter(old_path):
                entries.append({"path": old_path, "status": "D"})
            continue
        path = normalize_path(tokens[index])
        index += 1
        if path_filter(path):
            entries.append({"path": path, "status": status})
    return sorted(entries, key=lambda item: (item["path"], item["status"]))


def diff_entries(root: Path, arguments: Sequence[str]) -> list[dict[str, str]]:
    output = git(
        root,
        (*arguments, "--name-status", "--find-renames", "-z", "--", *GIT_SCOPES),
    )
    return parse_name_status(output)


def untracked_entries(root: Path) -> list[dict[str, str]]:
    output = git(
        root,
        ("ls-files", "--others", "--exclude-standard", "-z", "--", *GIT_SCOPES),
    )
    entries = []
    for token in output.split(b"\0"):
        if not token:
            continue
        path = normalize_path(decode_git_token(token))
        if is_relevant(path):
            entries.append({"path": path, "status": "??"})
    return sorted(entries, key=lambda item: item["path"])


def all_diff_entries(root: Path, arguments: Sequence[str]) -> list[dict[str, str]]:
    output = git(
        root,
        (*arguments, "--name-status", "--find-renames", "-z", "--"),
    )
    return parse_name_status(output, lambda path: bool(normalize_path(path)))


def all_untracked_entries(root: Path) -> list[dict[str, str]]:
    output = git(root, ("ls-files", "--others", "--exclude-standard", "-z", "--"))
    entries = []
    for token in output.split(b"\0"):
        if not token:
            continue
        path = normalize_path(decode_git_token(token))
        if path:
            entries.append({"path": path, "status": "??"})
    return sorted(entries, key=lambda item: item["path"])


def worktree_guard(root: Path) -> dict[str, Any]:
    groups = (
        ("staged", all_diff_entries(root, ("diff", "--cached"))),
        ("unstaged", all_diff_entries(root, ("diff",))),
        ("untracked", all_untracked_entries(root)),
    )
    states: dict[str, set[str]] = {}
    forbidden_paths: set[str] = set()
    for source, entries in groups:
        for entry in entries:
            paths = [entry["path"]]
            if entry.get("old_path"):
                paths.append(entry["old_path"])
            for path in paths:
                if is_forbidden(path):
                    forbidden_paths.add(path)
                    continue
                states.setdefault(path, set()).add(f"{source}:{entry['status']}")
    return {
        "files": [
            {"path": path, "states": sorted(values)}
            for path, values in sorted(states.items())
        ],
        "forbidden_change_count": len(forbidden_paths),
    }


def verification_error(
    code: str, message: str, suggestion: str, **details: Any
) -> dict[str, Any]:
    return {
        "code": code,
        "message": message,
        "suggestion": suggestion,
        **details,
    }


def compare_worktree_guards(
    before: dict[str, Any], current: dict[str, Any], allowed_documents: set[str]
) -> list[dict[str, Any]]:
    errors: list[dict[str, Any]] = []
    before_map = {item["path"]: item["states"] for item in before.get("files", [])}
    current_map = {item["path"]: item["states"] for item in current.get("files", [])}
    for path in sorted(set(before_map) | set(current_map)):
        if path in allowed_documents:
            continue
        if before_map.get(path) != current_map.get(path):
            errors.append(
                verification_error(
                    "write_outside_allowlist",
                    "检测到计划外文件状态发生变化。",
                    "撤销本流程对该文件的修改，或重新生成包含正确范围的新计划。",
                    path=path,
                )
            )
    if before.get("forbidden_change_count", 0) != current.get(
        "forbidden_change_count", 0
    ):
        errors.append(
            verification_error(
                "sensitive_or_excluded_path_changed",
                "检测到敏感或排除路径的工作区状态发生变化。",
                "停止文档同步并由用户单独检查敏感或排除文件。",
            )
        )
    return errors


def current_head(root: Path) -> str | None:
    output = git(root, ("rev-parse", "--verify", "HEAD"), required=False)
    value = output.decode("ascii", errors="ignore").strip().lower()
    return value if re.fullmatch(r"[0-9a-f]{40}", value) else None


def marker_is_usable(root: Path, marker: str, head: str | None) -> bool:
    if head is None:
        return False
    exists = subprocess.run(
        ["git", "cat-file", "-e", f"{marker}^{{commit}}"],
        cwd=root,
        env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"},
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    ).returncode == 0
    if not exists:
        return False
    return subprocess.run(
        ["git", "merge-base", "--is-ancestor", marker, head],
        cwd=root,
        env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"},
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    ).returncode == 0


def document_states(
    root: Path, warnings: set[str], documents: Sequence[str] = TARGET_DOCUMENTS
) -> tuple[dict[str, dict[str, Any]], dict[str, str | None]]:
    states: dict[str, dict[str, Any]] = {}
    markers: dict[str, str | None] = {}
    for path in documents:
        data = read_bytes(root, path, warnings)
        if data is None:
            states[path] = {"exists": False, "headings": [], "sha256": None}
            markers[path] = None
            warnings.add(f"target_document_missing:{path}")
            continue
        try:
            text = data.decode("utf-8-sig")
        except UnicodeDecodeError:
            text = ""
            warnings.add(f"target_document_not_utf8:{path}")
        marker_match = SYNC_MARKER_RE.search(text)
        marker = marker_match.group(1).lower() if marker_match else None
        markers[path] = marker
        headings = [
            {"level": len(match.group(1)), "title": match.group(2).strip()}
            for match in HEADING_RE.finditer(text)
        ]
        states[path] = {
            "exists": True,
            "headings": headings,
            "sha256": hashlib.sha256(data).hexdigest(),
            "sync_commit": marker,
        }
    return states, markers


def walk_relevant_files(root: Path) -> list[str]:
    roots = (
        "core/backend/airshelf",
        "core/backend/apps",
        "core/backend/requirements.txt",
        "core/backend/Dockerfile",
        "core/backend/docker-entrypoint.sh",
        "core/frontend/package.json",
        "core/frontend/src",
        "core/frontend/vite.config.ts",
        "core/frontend/Dockerfile",
        "core/frontend/nginx.conf",
        "k8s/core",
    )
    paths: set[str] = set()
    for relative in roots:
        candidate = safe_candidate(root, relative)
        if candidate is None or not candidate.exists():
            continue
        candidates: Iterable[Path] = (candidate,) if candidate.is_file() else candidate.rglob("*")
        for item in candidates:
            if not item.is_file() or item.is_symlink():
                continue
            try:
                rel = item.relative_to(root).as_posix()
            except ValueError:
                continue
            if is_relevant(rel):
                paths.add(rel)
    return sorted(paths)


def scope_content_fingerprint(
    root: Path, scope: str, warnings: set[str]
) -> tuple[str, int]:
    paths = [path for path in walk_relevant_files(root) if path_in_scope(path, scope)]
    fingerprints = {path: file_fingerprint(root, path, warnings) for path in paths}
    canonical = json.dumps(
        fingerprints, ensure_ascii=True, sort_keys=True, separators=(",", ":")
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:20], len(paths)


def parse_python(root: Path, relative: str, warnings: set[str]) -> ast.Module | None:
    text = read_text(root, relative, warnings)
    if text is None:
        return None
    try:
        return ast.parse(text, filename=relative)
    except SyntaxError:
        warnings.add(f"python_parse_failed:{relative}")
        return None


def dotted_name(node: ast.AST) -> str:
    if isinstance(node, ast.Name):
        return node.id
    if isinstance(node, ast.Attribute):
        parent = dotted_name(node.value)
        return f"{parent}.{node.attr}" if parent else node.attr
    return ""


def model_classes(root: Path, relative: str, warnings: set[str]) -> list[str]:
    tree = parse_python(root, relative, warnings)
    if tree is None:
        return []
    classes = []
    for node in tree.body:
        if not isinstance(node, ast.ClassDef):
            continue
        bases = [dotted_name(base).split(".")[-1] for base in node.bases]
        if any(base.lower().endswith("model") or base == "AbstractUser" for base in bases):
            classes.append(node.name)
    return sorted(classes)


def django_routes(root: Path, relative: str, warnings: set[str]) -> list[dict[str, str | None]]:
    tree = parse_python(root, relative, warnings)
    if tree is None:
        return []
    routes: list[dict[str, str | None]] = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call) or not node.args:
            continue
        if dotted_name(node.func).split(".")[-1] not in {"path", "re_path"}:
            continue
        first = node.args[0]
        if not isinstance(first, ast.Constant) or not isinstance(first.value, str):
            continue
        included: str | None = None
        if len(node.args) > 1 and isinstance(node.args[1], ast.Call):
            call = node.args[1]
            if dotted_name(call.func).split(".")[-1] == "include" and call.args:
                value = call.args[0]
                if isinstance(value, ast.Constant) and isinstance(value.value, str):
                    included = value.value
        routes.append({"include": included, "path": first.value})
    return sorted(routes, key=lambda item: (item["path"], item["include"] or ""))


def celery_tasks(root: Path, relative: str, warnings: set[str]) -> list[str]:
    tree = parse_python(root, relative, warnings)
    if tree is None:
        return []
    tasks = []
    for node in tree.body:
        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            continue
        decorator_names = []
        for decorator in node.decorator_list:
            target = decorator.func if isinstance(decorator, ast.Call) else decorator
            decorator_names.append(dotted_name(target).split(".")[-1])
        if any(name in {"shared_task", "task", "periodic_task"} for name in decorator_names):
            tasks.append(node.name)
    return sorted(tasks)


def top_level_classes(root: Path, relative: str, warnings: set[str]) -> list[str]:
    tree = parse_python(root, relative, warnings)
    if tree is None:
        return []
    return sorted(node.name for node in tree.body if isinstance(node, ast.ClassDef))


def assignment_names(tree: ast.Module) -> set[str]:
    names: set[str] = set()
    for node in tree.body:
        targets: list[ast.AST] = []
        if isinstance(node, ast.Assign):
            targets.extend(node.targets)
        elif isinstance(node, ast.AnnAssign):
            targets.append(node.target)
        for target in targets:
            if isinstance(target, ast.Name):
                names.add(target.id)
    return names


def requirement_names(root: Path, warnings: set[str]) -> list[str]:
    text = read_text(root, "core/backend/requirements.txt", warnings)
    if text is None:
        return []
    names = set()
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith(("#", "-")):
            continue
        match = re.match(r"([A-Za-z0-9_.-]+)", stripped)
        if match:
            names.add(match.group(1).lower())
    return sorted(names)


def collect_backend(root: Path, warnings: set[str]) -> dict[str, Any]:
    apps_root = root / "core" / "backend" / "apps"
    apps = []
    if apps_root.is_dir() and not apps_root.is_symlink():
        apps = sorted(
            item.name
            for item in apps_root.iterdir()
            if item.is_dir()
            and not item.is_symlink()
            and not is_forbidden(f"core/backend/apps/{item.name}")
            and ((item / "apps.py").is_file() or (item / "__init__.py").is_file())
        )

    models: dict[str, list[str]] = {}
    app_routes: dict[str, list[dict[str, str | None]]] = {}
    tasks: dict[str, list[str]] = {}
    for app in apps:
        model_path = f"core/backend/apps/{app}/models.py"
        route_path = f"core/backend/apps/{app}/urls.py"
        task_path = f"core/backend/apps/{app}/tasks.py"
        if safe_candidate(root, model_path) and (root / model_path).is_file():
            models[app] = model_classes(root, model_path, warnings)
        if safe_candidate(root, route_path) and (root / route_path).is_file():
            app_routes[app] = django_routes(root, route_path, warnings)
        if safe_candidate(root, task_path) and (root / task_path).is_file():
            tasks[app] = celery_tasks(root, task_path, warnings)

    provider_root = root / "core" / "backend" / "apps" / "ai" / "providers"
    providers = []
    if provider_root.is_dir() and not provider_root.is_symlink():
        for file in sorted(provider_root.glob("*.py"), key=lambda item: item.name):
            if file.name == "__init__.py" or file.is_symlink():
                continue
            relative = file.relative_to(root).as_posix()
            providers.append(
                {"classes": top_level_classes(root, relative, warnings), "path": relative}
            )

    settings = {}
    settings_root = root / "core" / "backend" / "airshelf" / "settings"
    if settings_root.is_dir() and not settings_root.is_symlink():
        important = {
            "AUTHENTICATION_BACKENDS",
            "AUTH_USER_MODEL",
            "CACHES",
            "DATABASES",
            "INSTALLED_APPS",
            "MIDDLEWARE",
            "REST_FRAMEWORK",
            "STORAGES",
        }
        prefixes = ("AI_", "ARK_", "CELERY_", "MODEL_", "REDIS_", "TOKENSSR_", "TOS_", "YUNQI_")
        for file in sorted(settings_root.glob("*.py"), key=lambda item: item.name):
            if file.name == "__init__.py" or file.is_symlink():
                continue
            relative = file.relative_to(root).as_posix()
            tree = parse_python(root, relative, warnings)
            symbols = [] if tree is None else sorted(
                name for name in assignment_names(tree) if name in important or name.startswith(prefixes)
            )
            settings[file.stem] = symbols

    base_settings = read_text(root, "core/backend/airshelf/settings/base.py", warnings) or ""
    configured_apps = sorted(set(re.findall(r"[\"']apps\.([a-zA-Z0-9_]+)", base_settings)))
    deployment_candidates = (
        "core/backend/Dockerfile",
        "core/backend/docker-entrypoint.sh",
    )
    deployment_files = [
        path for path in deployment_candidates if (safe_candidate(root, path) or Path()).is_file()
    ]
    k8s_files = [path for path in walk_relevant_files(root) if path.startswith("k8s/core/")]
    return {
        "api_routes": django_routes(root, "core/backend/airshelf/urls.py", warnings),
        "app_routes": app_routes,
        "apps": apps,
        "celery_tasks": tasks,
        "configured_apps": configured_apps,
        "deployment_files": deployment_files + k8s_files,
        "models": models,
        "providers": providers,
        "requirements": requirement_names(root, warnings),
        "settings_symbols": settings,
    }


def relative_source_files(root: Path, base: str, suffixes: tuple[str, ...]) -> list[str]:
    directory = safe_candidate(root, base)
    if directory is None or not directory.is_dir() or directory.is_symlink():
        return []
    files = []
    for item in directory.rglob("*"):
        if not item.is_file() or item.is_symlink() or item.suffix.lower() not in suffixes:
            continue
        relative = item.relative_to(root).as_posix()
        if not is_forbidden(relative):
            files.append(relative)
    return sorted(files)


def typescript_exports(root: Path, relative: str, warnings: set[str]) -> list[str]:
    text = read_text(root, relative, warnings)
    return [] if text is None else sorted(set(TS_EXPORT_RE.findall(text)))


def quoted_paths(root: Path, relative: str, warnings: set[str]) -> list[str]:
    text = read_text(root, relative, warnings)
    return [] if text is None else sorted(set(QUOTED_PATH_RE.findall(text)))


def stage_order(root: Path, warnings: set[str]) -> list[str]:
    text = read_text(root, "core/frontend/src/routes/stage-config.ts", warnings)
    if text is None:
        return []
    match = re.search(r"\bstageOrder\s*=\s*\[([^\]]*)\]", text, re.DOTALL)
    if not match:
        return []
    return re.findall(r"[\"']([A-Za-z0-9_-]+)[\"']", match.group(1))


def frontend_dependencies(root: Path, warnings: set[str]) -> dict[str, list[str]]:
    text = read_text(root, "core/frontend/package.json", warnings)
    if text is None:
        return {"dependencies": [], "devDependencies": []}
    try:
        package = json.loads(text)
    except json.JSONDecodeError:
        warnings.add("json_parse_failed:core/frontend/package.json")
        return {"dependencies": [], "devDependencies": []}
    return {
        group: sorted(str(name) for name in package.get(group, {}).keys())
        for group in ("dependencies", "devDependencies")
    }


def collect_frontend(root: Path, warnings: set[str]) -> dict[str, Any]:
    deployment_candidates = (
        "core/frontend/Dockerfile",
        "core/frontend/nginx.conf",
        "core/frontend/vite.config.ts",
    )
    deployment_files = [
        path for path in deployment_candidates if (safe_candidate(root, path) or Path()).is_file()
    ]
    return {
        "api_endpoints": quoted_paths(root, "core/frontend/src/api.ts", warnings),
        "api_exports": typescript_exports(root, "core/frontend/src/api.ts", warnings),
        "component_files": relative_source_files(
            root, "core/frontend/src/components", (".ts", ".tsx")
        ),
        "dependencies": frontend_dependencies(root, warnings),
        "deployment_files": deployment_files,
        "route_files": relative_source_files(root, "core/frontend/src/routes", (".ts", ".tsx")),
        "route_paths": quoted_paths(root, "core/frontend/src/routes/route-config.ts", warnings),
        "stage_order": stage_order(root, warnings),
        "type_exports": typescript_exports(root, "core/frontend/src/types.ts", warnings),
    }


def changed_line_text(root: Path, base: str, path: str, warnings: set[str]) -> list[str]:
    if not is_relevant(path):
        return []
    output = git(
        root,
        ("diff", "--no-ext-diff", "--unified=0", base, "--", path),
        required=False,
    )
    if not output:
        current = read_text(root, path, warnings)
        return [] if current is None else [f"+{line}" for line in current.splitlines()]
    if len(output) > 2_000_000:
        warnings.add(f"diff_too_large:{path}")
        output = output[:2_000_000]
    text = output.decode("utf-8", errors="replace")
    return [
        line
        for line in text.splitlines()
        if line.startswith(("+", "-")) and not line.startswith(("+++", "---"))
    ]


def text_at_commit(root: Path, commit: str, path: str, warnings: set[str]) -> str | None:
    if not is_relevant(path):
        return None
    output = git(root, ("show", f"{commit}:{path}"), required=False)
    if not output:
        return None
    if len(output) > MAX_TEXT_BYTES:
        warnings.add(f"base_file_too_large:{path}")
        return None
    try:
        return output.decode("utf-8-sig")
    except UnicodeDecodeError:
        warnings.add(f"base_file_not_utf8:{path}")
        return None


def change_index(entries: Iterable[dict[str, str]]) -> dict[str, set[str]]:
    indexed: dict[str, set[str]] = {}
    for entry in entries:
        status = entry["status"]
        code = "A" if status == "??" else status[:1]
        path = entry["path"]
        if path:
            indexed.setdefault(path, set()).add(code)
        old_path = entry.get("old_path")
        if old_path:
            indexed.setdefault(old_path, set()).add("D")
            indexed[path].add("A")
    return indexed


def dependency_names_from_package(text: str | None) -> set[str] | None:
    if text is None:
        return set()
    try:
        package = json.loads(text)
    except json.JSONDecodeError:
        return None
    names: set[str] = set()
    for group in ("dependencies", "devDependencies"):
        values = package.get(group, {})
        if isinstance(values, dict):
            names.update(str(name) for name in values)
    return names


def dependency_names_from_requirements(text: str | None) -> set[str]:
    if text is None:
        return set()
    names = set()
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith(("#", "-")):
            continue
        match = re.match(r"([A-Za-z0-9_.-]+)", stripped)
        if match:
            names.add(match.group(1).lower())
    return names


def dependency_set_changed(
    root: Path, base: str, path: str, warnings: set[str]
) -> bool:
    before = text_at_commit(root, base, path, warnings)
    after = read_text(root, path, warnings)
    if path.endswith("package.json"):
        before_names = dependency_names_from_package(before)
        after_names = dependency_names_from_package(after)
        if before_names is None or after_names is None:
            warnings.add(f"dependency_parse_failed:{path}")
            return True
        return before_names != after_names
    return dependency_names_from_requirements(before) != dependency_names_from_requirements(after)


def stage_order_from_text(text: str | None) -> list[str]:
    if text is None:
        return []
    match = re.search(r"\bstageOrder\s*=\s*\[([^\]]*)\]", text, re.DOTALL)
    if not match:
        return []
    return re.findall(r"[\"']([A-Za-z0-9_-]+)[\"']", match.group(1))


def stage_order_changed(root: Path, base: str, warnings: set[str]) -> bool:
    path = "core/frontend/src/routes/stage-config.ts"
    before = text_at_commit(root, base, path, warnings)
    after = read_text(root, path, warnings)
    return stage_order_from_text(before) != stage_order_from_text(after)


def frontend_page_section(path: str) -> str:
    if "/admin/" in path:
        return "8.7 平台超管后台"
    stem = PurePosixPath(path).stem
    sections = {
        "products": "8.1 商品",
        "product-detail": "8.1 商品",
        "product-create": "8.1 商品",
        "projects": "8.2 视频项目",
        "pipeline": "8.3 生产管线",
        "ai-tools": "8.4 图片创作",
        "models": "8.4 图片创作",
        "free-create": "8.5 自由视频创作",
        "library": "8.6 资产库与垃圾桶",
        "trash": "8.6 资产库与垃圾桶",
    }
    return sections.get(stem, "8. 页面模块")


def impact_spec(
    path: str,
    statuses: set[str],
    changed_lines: Sequence[str],
    *,
    dependency_changed: bool = False,
    stages_changed: bool = False,
) -> dict[str, str] | None:
    changed = "\n".join(changed_lines)
    added_or_deleted = bool(statuses & {"A", "D", "R", "C"})

    if path.startswith("core/backend/") or path.startswith("k8s/core/"):
        document = BACKEND_DOCUMENT
        app_match = re.match(r"core/backend/apps/([^/]+)/([^/]+\.py)$", path)
        app = app_match.group(1) if app_match else ""
        filename = app_match.group(2) if app_match else ""
        app_section = BACKEND_APP_SECTIONS.get(app, "2. 目录结构、4. 应用模块")

        if path.startswith("core/backend/airshelf/settings/") and re.search(
            r"(?i)(INSTALLED_APPS|MIDDLEWARE|DATABASE|DB_ENGINE|\bENGINE\b|CACHE|REDIS|CELERY|"
            r"REST_FRAMEWORK|AUTH_|Authentication|Permission|STORAGES|TOS_|ARK_|"
            r"YUNQI_|TOKENSSR_|MODEL_|AI_|apps\.)",
            changed,
        ):
            return {
                "change_type": "backend.settings_boundary",
                "classification": "architecture",
                "document": document,
                "reason": "后端框架、基础设施或鉴权配置边界发生变化",
                "section": "3.1 Settings、3.3 鉴权",
            }
        if path == "core/backend/airshelf/urls.py" and (
            added_or_deleted or re.search(r"\b(path|re_path|include)\s*\(", changed)
        ):
            return {
                "change_type": "backend.root_routes",
                "classification": "architecture",
                "document": document,
                "reason": "后端一级路由入口发生变化",
                "section": "3.2 路由",
            }
        if filename == "apps.py" and (
            added_or_deleted or re.search(r"(?m)^[+-].*\bname\s*=", changed)
        ):
            return {
                "change_type": "backend.django_apps",
                "classification": "architecture",
                "document": document,
                "reason": f"Django 应用 `{app}` 的注册结构发生变化",
                "section": "2. 目录结构、4. 应用模块",
            }
        if filename == "models.py" and (
            added_or_deleted
            or re.search(r"(?m)^[+-]\s*class\s+\w+\s*\([^\n]*(?:Model|AbstractUser)", changed)
        ):
            return {
                "change_type": "backend.model_classes",
                "classification": "architecture",
                "document": document,
                "reason": f"`{app}` 核心模型类型发生增删或继承变化",
                "section": app_section,
            }
        if filename == "urls.py" and (
            added_or_deleted or re.search(r"\b(path|re_path|include)\s*\(", changed)
        ):
            return {
                "change_type": "backend.app_routes",
                "classification": "architecture",
                "document": document,
                "reason": f"`{app}` 应用接口边界发生变化",
                "section": app_section,
            }
        if filename == "tasks.py" and (
            added_or_deleted or re.search(r"(shared_task|periodic_task|\.task\s*\()", changed)
        ):
            return {
                "change_type": "backend.celery_tasks",
                "classification": "architecture",
                "document": document,
                "reason": f"`{app}` 的 Celery 任务入口发生变化",
                "section": "7.1 Celery",
            }
        if path.startswith("core/backend/apps/ai/providers/") and (
            added_or_deleted
            or re.search(r"(?m)^[+-]\s*class\s+", changed)
            or (
                PurePosixPath(path).name == "__init__.py"
                and re.search(r"(?m)^[+-]\s*(?:from|import)\s+", changed)
            )
        ):
            return {
                "change_type": "backend.ai_providers",
                "classification": "architecture",
                "document": document,
                "reason": "AI Provider 类型或适配层结构发生变化",
                "section": "6. AI Provider 架构",
            }
        if path == "core/backend/requirements.txt" and dependency_changed:
            return {
                "change_type": "backend.dependencies",
                "classification": "architecture",
                "document": document,
                "reason": "后端依赖集合发生变化",
                "section": "10. 部署",
            }
        if path in {"core/backend/Dockerfile", "core/backend/docker-entrypoint.sh"} or path.startswith(
            "k8s/core/"
        ):
            if changed_lines or added_or_deleted:
                return {
                    "change_type": "backend.deployment",
                    "classification": "architecture",
                    "document": document,
                    "reason": "后端容器或部署拓扑发生变化",
                    "section": "10. 部署",
                }
        return None

    if path.startswith("core/frontend/"):
        document = FRONTEND_DOCUMENT
        if path == "core/frontend/package.json" and dependency_changed:
            return {
                "change_type": "frontend.dependencies",
                "classification": "architecture",
                "document": document,
                "reason": "前端依赖集合发生变化",
                "section": "2. 技术栈",
            }
        if path == "core/frontend/src/routes/route-config.ts" and (
            added_or_deleted
            or re.search(r"(?m)^[+-].*(?:[\"'`]/|type\s+Page|mainNav|supplementNav|pathForPage)", changed)
        ):
            return {
                "change_type": "frontend.routing",
                "classification": "architecture",
                "document": document,
                "reason": "前端页面路由或导航边界发生变化",
                "section": "5. 路由架构",
            }
        if path == "core/frontend/src/routes/stage-config.ts" and stages_changed:
            return {
                "change_type": "frontend.pipeline_stages",
                "classification": "architecture",
                "document": document,
                "reason": "五阶段顺序或阶段标识发生变化",
                "section": "8.3 生产管线",
            }
        if path.startswith("core/frontend/src/routes/") and PurePosixPath(path).suffix in {
            ".ts",
            ".tsx",
        } and added_or_deleted:
            return {
                "change_type": "frontend.pages",
                "classification": "architecture",
                "document": document,
                "reason": "正式业务页面模块发生增删或移动",
                "section": frontend_page_section(path),
            }
        if path == "core/frontend/src/App.tsx" and re.search(
            r"(?i)(createContext|Context\.Provider|useReducer|redux|zustand|RouterProvider|BrowserRouter)",
            changed,
        ):
            return {
                "change_type": "frontend.app_controller",
                "classification": "architecture",
                "document": document,
                "reason": "应用总控制器或全局状态边界发生变化",
                "section": "4.2 `App.tsx`",
            }
        if path == "core/frontend/src/api.ts" and re.search(
            r"(?i)(VITE_API_BASE_URL|Authorization|localStorage|sessionStorage|fetch\s*\(|"
            r"axios|(?:const|function)\s+request\b|export\s+const\s+\w*Api\b)",
            changed,
        ):
            return {
                "change_type": "frontend.api_boundary",
                "classification": "architecture",
                "document": document,
                "reason": "前端请求、鉴权或 API SDK 边界发生变化",
                "section": "6. API 层、7. 登录、存储与权限",
            }
        if path.startswith("core/frontend/src/components/") and added_or_deleted and re.search(
            r"(?i)(shell|layout|router|provider|store|context)", PurePosixPath(path).stem
        ):
            return {
                "change_type": "frontend.shared_infrastructure",
                "classification": "architecture",
                "document": document,
                "reason": "前端共享基础组件发生增删或移动",
                "section": "3. 目录结构、4. 应用入口",
            }
        if path in {
            "core/frontend/Dockerfile",
            "core/frontend/nginx.conf",
            "core/frontend/vite.config.ts",
        } and (changed_lines or added_or_deleted):
            return {
                "change_type": "frontend.deployment",
                "classification": "architecture",
                "document": document,
                "reason": "前端构建、代理或容器部署边界发生变化",
                "section": "11. 构建与部署",
            }
    return None


def baseline_impacts(document: str, baseline_files: Sequence[str]) -> list[dict[str, Any]]:
    if document == BACKEND_DOCUMENT:
        groups = (
            (
                "backend.baseline.apps",
                "2. 目录结构、4. 应用模块",
                "缺少同步标记，需要核对 Django 应用与核心模型基线",
                lambda path: "/apps/" in path and path.endswith(("apps.py", "models.py", "urls.py")),
            ),
            (
                "backend.baseline.entry",
                "3. Django 工程入口",
                "缺少同步标记，需要核对 Settings 与一级路由基线",
                lambda path: "/airshelf/settings/" in path or path.endswith("airshelf/urls.py"),
            ),
            (
                "backend.baseline.execution",
                "6. AI Provider 架构、7. 任务执行模型",
                "缺少同步标记，需要核对 Provider 与 Celery 任务基线",
                lambda path: "/providers/" in path or path.endswith("tasks.py"),
            ),
            (
                "backend.baseline.deployment",
                "10. 部署",
                "缺少同步标记，需要核对依赖、容器与部署基线",
                lambda path: path.endswith(("requirements.txt", "Dockerfile", "docker-entrypoint.sh"))
                or path.startswith("k8s/core/"),
            ),
        )
    else:
        groups = (
            (
                "frontend.baseline.stack",
                "2. 技术栈",
                "缺少同步标记，需要核对前端技术依赖基线",
                lambda path: path.endswith("package.json"),
            ),
            (
                "frontend.baseline.structure",
                "3. 目录结构、5. 路由架构、8. 页面模块",
                "缺少同步标记，需要核对页面、路由与共享组件基线",
                lambda path: "/src/routes/" in path or "/src/components/" in path,
            ),
            (
                "frontend.baseline.boundaries",
                "4. 应用入口、6. API 层",
                "缺少同步标记，需要核对应用入口与 API SDK 基线",
                lambda path: path.endswith(("src/App.tsx", "src/api.ts", "src/types.ts")),
            ),
            (
                "frontend.baseline.deployment",
                "11. 构建与部署",
                "缺少同步标记，需要核对构建、代理与容器基线",
                lambda path: path.endswith(("Dockerfile", "nginx.conf", "vite.config.ts")),
            ),
        )
    impacts = []
    for change_type, section, reason, matches in groups:
        evidence = sorted(path for path in baseline_files if matches(path))[:24]
        if evidence:
            impacts.append(
                {
                    "change_type": change_type,
                    "classification": "baseline_review",
                    "document": document,
                    "evidence": evidence,
                    "reason": reason,
                    "section": section,
                }
            )
    return impacts


def map_impacts(
    root: Path,
    head: str | None,
    documents: Sequence[str],
    usable_markers: dict[str, str],
    baseline_files: Sequence[str],
    staged: Sequence[dict[str, str]],
    unstaged: Sequence[dict[str, str]],
    untracked: Sequence[dict[str, str]],
    warnings: set[str],
) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
    impacts: list[dict[str, Any]] = []
    ordinary: dict[str, str] = {}
    for document in documents:
        base = usable_markers.get(document)
        if base is None or head is None:
            impacts.extend(baseline_impacts(document, baseline_files))
            continue
        committed = diff_entries(root, ("diff", f"{base}..{head}"))
        indexed = change_index([*committed, *staged, *unstaged, *untracked])
        if document == BACKEND_DOCUMENT:
            domain_paths = sorted(
                path
                for path in indexed
                if path.startswith("core/backend/") or path.startswith("k8s/core/")
            )
        else:
            domain_paths = sorted(
                path for path in indexed if path.startswith("core/frontend/")
            )
        for path in domain_paths:
            lines = changed_line_text(root, base, path, warnings)
            dependency_changed = path in {
                "core/backend/requirements.txt",
                "core/frontend/package.json",
            } and dependency_set_changed(root, base, path, warnings)
            stages_changed = path == "core/frontend/src/routes/stage-config.ts" and stage_order_changed(
                root, base, warnings
            )
            spec = impact_spec(
                path,
                indexed[path],
                lines,
                dependency_changed=dependency_changed,
                stages_changed=stages_changed,
            )
            if spec is None:
                ordinary[path] = "未发现会改变架构文档事实的结构信号"
                continue
            impacts.append({**spec, "evidence": [path]})

    buckets: dict[tuple[str, str, str, str, str], set[str]] = {}
    for impact in impacts:
        key = (
            impact["document"],
            impact["section"],
            impact["change_type"],
            impact["classification"],
            impact["reason"],
        )
        buckets.setdefault(key, set()).update(impact["evidence"])
    merged = [
        {
            "change_type": key[2],
            "classification": key[3],
            "document": key[0],
            "evidence": sorted(evidence),
            "reason": key[4],
            "section": key[1],
        }
        for key, evidence in buckets.items()
    ]
    merged.sort(key=lambda item: (item["document"], item["section"], item["change_type"]))
    non_architecture = [
        {"path": path, "reason": reason} for path, reason in sorted(ordinary.items())
    ]
    return merged, non_architecture


def encode_guard(payload: dict[str, Any]) -> str:
    raw = json.dumps(payload, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode(
        "utf-8"
    )
    encoded = base64.urlsafe_b64encode(zlib.compress(raw, level=9)).decode("ascii").rstrip("=")
    checksum = hashlib.sha256(raw).hexdigest()[:16]
    return f"{encoded}.{checksum}"


def decode_guard(token: str) -> dict[str, Any]:
    try:
        encoded, checksum = token.rsplit(".", 1)
        padding = "=" * (-len(encoded) % 4)
        raw = zlib.decompress(base64.urlsafe_b64decode(encoded + padding))
        if hashlib.sha256(raw).hexdigest()[:16] != checksum:
            raise ValueError("checksum")
        payload = json.loads(raw.decode("utf-8"))
    except (ValueError, UnicodeDecodeError, json.JSONDecodeError, zlib.error) as error:
        raise ScanError("verification_guard_invalid") from error
    if not isinstance(payload, dict) or payload.get("guard_version") != 1:
        raise ScanError("verification_guard_invalid")
    return payload


def headings_from_text(text: str, maximum_level: int = 3) -> list[dict[str, Any]]:
    return [
        {"level": len(match.group(1)), "title": match.group(2).strip()}
        for match in HEADING_RE.finditer(text)
        if len(match.group(1)) <= maximum_level
    ]


def section_records(text: str) -> list[dict[str, Any]]:
    matches = [
        match for match in HEADING_RE.finditer(text) if len(match.group(1)) <= 3
    ]
    records: list[dict[str, Any]] = []
    stack: list[tuple[int, str]] = []
    key_counts: dict[str, int] = {}
    for index, match in enumerate(matches):
        level = len(match.group(1))
        title = match.group(2).strip()
        while stack and stack[-1][0] >= level:
            stack.pop()
        ancestors = [item[1] for item in stack]
        stack.append((level, title))
        end = len(text)
        for candidate in matches[index + 1 :]:
            if len(candidate.group(1)) <= level:
                end = candidate.start()
                break
        base_key = f"{level}:{title}"
        key_counts[base_key] = key_counts.get(base_key, 0) + 1
        key = (
            base_key
            if key_counts[base_key] == 1
            else f"{base_key}#{key_counts[base_key]}"
        )
        records.append(
            {
                "ancestors": ancestors,
                "end": end,
                "hash": hashlib.sha256(text[match.start() : end].encode("utf-8")).hexdigest(),
                "key": key,
                "level": level,
                "start": match.start(),
                "title": title,
            }
        )
    return records


def protected_section_hashes(text: str, planned_sections: Sequence[str]) -> dict[str, str]:
    records = section_records(text)
    planned = set(planned_sections)
    protected: dict[str, str] = {}
    for record in records:
        descendant_planned = any(
            other["title"] in planned
            and other["start"] > record["start"]
            and other["start"] < record["end"]
            for other in records
        )
        affected = (
            record["title"] in planned
            or any(ancestor in planned for ancestor in record["ancestors"])
            or descendant_planned
        )
        if not affected:
            protected[record["key"]] = record["hash"]
    return protected


def document_header_hash(text: str) -> str:
    second_level = re.search(r"(?m)^##\s+", text)
    header = text[: second_level.start()] if second_level else text
    kept_lines = []
    for line in header.splitlines():
        stripped = line.strip()
        if re.fullmatch(r">\s*最后核对：\d{4}-\d{2}-\d{2}。", stripped):
            continue
        if SYNC_MARKER_RE.fullmatch(stripped):
            continue
        kept_lines.append(line.rstrip())
    normalized = "\n".join(kept_lines).strip()
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()


def is_ordered_subsequence(before: Sequence[str], current: Sequence[str]) -> bool:
    iterator = iter(current)
    return all(any(candidate == expected for candidate in iterator) for expected in before)


def validate_headings(
    document: str,
    before: Sequence[dict[str, Any]],
    current: Sequence[dict[str, Any]],
) -> list[dict[str, Any]]:
    errors: list[dict[str, Any]] = []
    before_primary = [item["title"] for item in before if item["level"] <= 2]
    current_primary = [item["title"] for item in current if item["level"] <= 2]
    if before_primary != current_primary:
        errors.append(
            verification_error(
                "primary_headings_changed",
                "原有一级或二级标题被删除、改名、增加或调整顺序。",
                "恢复原有一级和二级标题；内容变化应限制在已批准章节正文内。",
                document=document,
            )
        )
    before_tertiary = [item["title"] for item in before if item["level"] == 3]
    current_tertiary = [item["title"] for item in current if item["level"] == 3]
    if not is_ordered_subsequence(before_tertiary, current_tertiary):
        errors.append(
            verification_error(
                "existing_subheadings_removed_or_reordered",
                "原有三级标题被删除或调整顺序。",
                "恢复原有三级标题及顺序；如需新增标题，只能追加到已批准章节。",
                document=document,
            )
        )
    return errors


def markdown_targets(text: str) -> list[str]:
    values = MARKDOWN_LINK_RE.findall(text) + MARKDOWN_REFERENCE_LINK_RE.findall(text)
    targets = []
    for value in values:
        target = value.strip().strip("<>")
        if " " in target and not target.startswith(("http://", "https://")):
            target = target.split(" ", 1)[0]
        targets.append(target)
    return targets


def validate_relative_links(root: Path, document: str, text: str) -> list[dict[str, Any]]:
    errors: list[dict[str, Any]] = []
    document_path = root.joinpath(*PurePosixPath(document).parts)
    for raw_target in markdown_targets(text):
        if not raw_target or raw_target.startswith("#"):
            continue
        if raw_target.lower().startswith("file://"):
            errors.append(
                verification_error(
                    "absolute_link_forbidden",
                    "文档包含本机绝对链接。",
                    "改为仓库内相对链接。",
                    document=document,
                )
            )
            continue
        target = unquote(raw_target.split("#", 1)[0].split("?", 1)[0])
        if not target:
            continue
        if target.startswith(("/", "\\")) or re.match(r"^[A-Za-z]:[\\/]", target):
            errors.append(
                verification_error(
                    "absolute_link_forbidden",
                    "文档包含本机或根路径链接。",
                    "改为仓库内相对链接。",
                    document=document,
                )
            )
            continue
        if re.match(r"^[A-Za-z][A-Za-z0-9+.-]*:", raw_target):
            continue
        candidate = (document_path.parent / target).resolve(strict=False)
        try:
            relative = candidate.relative_to(root.resolve(strict=True)).as_posix()
        except (OSError, ValueError):
            errors.append(
                verification_error(
                    "link_outside_repository",
                    "相对链接解析到了仓库外部。",
                    "将链接改为仓库内部存在的相对路径。",
                    document=document,
                )
            )
            continue
        if is_forbidden(relative) or not candidate.exists():
            errors.append(
                verification_error(
                    "relative_link_invalid",
                    "文档包含不存在或不允许引用的相对链接。",
                    "修正链接目标，或移除无法由仓库内容证明的链接。",
                    document=document,
                    target=raw_target,
                )
            )
    return errors


def validate_document_safety(document: str, text: str) -> list[dict[str, Any]]:
    errors: list[dict[str, Any]] = []
    if ABSOLUTE_LOCAL_PATH_RE.search(text):
        errors.append(
            verification_error(
                "absolute_local_path_detected",
                "文档包含本机绝对路径。",
                "删除本机路径并改用仓库相对路径。",
                document=document,
            )
        )
    if any(pattern.search(text) for pattern in SECRET_CONTENT_PATTERNS):
        errors.append(
            verification_error(
                "sensitive_content_detected",
                "文档包含疑似密钥、密码或私钥内容。",
                "立即删除敏感值，仅保留环境变量名称或安全配置说明。",
                document=document,
            )
        )
    return errors


def referenced_code_paths(document: str, text: str) -> list[str]:
    candidates = set()
    for value in re.findall(r"`([^`\n]+)`", text):
        value = value.strip().replace("\\", "/")
        if not value or any(char in value for char in "*?{}<>") or " " in value:
            continue
        if value.startswith("core/"):
            candidate = value.rstrip("/.,;:")
        elif document == BACKEND_DOCUMENT and value.startswith(("apps/", "airshelf/", "skills/")):
            candidate = f"core/backend/{value}".rstrip("/.,;:")
        elif document == FRONTEND_DOCUMENT and value.startswith(("src/", "public/")):
            candidate = f"core/frontend/{value}".rstrip("/.,;:")
        else:
            continue
        candidates.add(normalize_path(candidate))
    return sorted(path for path in candidates if path)


def validate_code_references(
    root: Path, document: str, text: str
) -> tuple[list[dict[str, Any]], list[str]]:
    errors: list[dict[str, Any]] = []
    checked: list[str] = []
    for path in referenced_code_paths(document, text):
        if is_forbidden(path):
            errors.append(
                verification_error(
                    "forbidden_code_reference",
                    "文档引用了敏感或排除路径。",
                    "移除该引用，仅保留允许公开的架构事实。",
                    document=document,
                )
            )
            continue
        candidate = safe_candidate(root, path)
        if candidate is None or not candidate.exists():
            errors.append(
                verification_error(
                    "code_reference_missing",
                    "文档中的代码路径无法在仓库中验证。",
                    "修正路径，或删除没有代码依据的描述。",
                    document=document,
                    path=path,
                )
            )
            continue
        checked.append(path)
    return errors, checked


def expand_sections(section: str) -> list[str]:
    return [part.strip() for part in section.split("、") if part.strip()]


def planned_updates(impacts: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
    sections_by_document: dict[str, set[str]] = {}
    for impact in impacts:
        sections_by_document.setdefault(impact["document"], set()).update(
            expand_sections(impact["section"])
        )
    return [
        {"document": document, "sections": sorted(sections)}
        for document, sections in sorted(sections_by_document.items())
    ]


def build_snapshot(root: Path, mode: str, scope: str = "all") -> dict[str, Any]:
    warnings: set[str] = set()
    targets = selected_documents(scope)
    head = current_head(root)
    if head is None:
        warnings.add("head_commit_unavailable")

    documents, markers = document_states(root, warnings, targets)
    usable_markers: dict[str, str] = {}
    for document, marker in markers.items():
        if marker is None:
            warnings.add(f"sync_marker_missing:{document}")
        elif marker_is_usable(root, marker, head):
            usable_markers[document] = marker
        else:
            warnings.add(f"sync_marker_unusable:{document}")

    baseline_required = len(usable_markers) != len(targets)
    committed = []
    for marker in sorted(set(usable_markers.values())):
        committed.append(
            {
                "base_commit": marker,
                "files": filter_entries(
                    diff_entries(root, ("diff", f"{marker}..{head}")), scope
                ),
            }
        )
    baseline_files = (
        [path for path in walk_relevant_files(root) if path_in_scope(path, scope)]
        if baseline_required
        else []
    )
    staged = filter_entries(diff_entries(root, ("diff", "--cached")), scope)
    unstaged = filter_entries(diff_entries(root, ("diff",)), scope)
    untracked = filter_entries(untracked_entries(root), scope)

    changed_paths = set(baseline_files)
    for group in committed:
        changed_paths.update(item["path"] for item in group["files"])
    for group in (staged, unstaged, untracked):
        changed_paths.update(item["path"] for item in group)

    common_marker: str | None = None
    marker_values = set(usable_markers.values())
    if len(usable_markers) == len(targets) and len(marker_values) == 1:
        common_marker = next(iter(marker_values))

    impacts, non_architecture = map_impacts(
        root,
        head,
        targets,
        usable_markers,
        baseline_files,
        staged,
        unstaged,
        untracked,
        warnings,
    )
    architecture: dict[str, Any] = {}
    if BACKEND_DOCUMENT in targets:
        architecture["backend"] = collect_backend(root, warnings)
    if FRONTEND_DOCUMENT in targets:
        architecture["frontend"] = collect_frontend(root, warnings)
    content_fingerprint, fingerprinted_file_count = scope_content_fingerprint(
        root, scope, warnings
    )
    snapshot_core = {
        "architecture": architecture,
        "changed_files": sorted(changed_paths),
        "changes": {
            "baseline": baseline_files,
            "committed": committed,
            "staged": staged,
            "unstaged": unstaged,
            "untracked": untracked,
        },
        "content_fingerprint": content_fingerprint,
        "documents": documents,
        "fingerprinted_file_count": fingerprinted_file_count,
        "repository": {
            "base_commit": common_marker,
            "baseline_documents": sorted(set(targets) - set(usable_markers)),
            "baseline_required": baseline_required,
            "current_commit": head,
            "sync_commits": markers,
        },
        "scope": scope,
    }
    canonical = json.dumps(snapshot_core, ensure_ascii=True, sort_keys=True, separators=(",", ":"))
    snapshot_id = hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:20]
    warnings_list = sorted(warnings)
    update_plan = planned_updates(impacts)
    requires_update = bool(impacts)
    plan_material = {
        "impacts": impacts,
        "non_architecture_changes": non_architecture,
        "planned_updates": update_plan,
        "scope": scope,
        "snapshot_id": snapshot_id,
        "warnings": warnings_list,
    }
    plan_canonical = json.dumps(
        plan_material, ensure_ascii=True, sort_keys=True, separators=(",", ":")
    )
    plan_id = f"arch-{hashlib.sha256(plan_canonical.encode('utf-8')).hexdigest()[:20]}"
    evidence_allowlist = sorted(
        {path for impact in impacts for path in impact.get("evidence", [])}
    )
    document_preconditions = [
        {
            "document": item["document"],
            "before_sha256": documents[item["document"]]["sha256"],
            "sections": item["sections"],
        }
        for item in update_plan
    ]
    working_tree_evidence = sorted(
        {
            entry["path"]
            for entries in (staged, unstaged, untracked)
            for entry in entries
        }
    )
    control_paths = (
        "tools/architecture-sync/WORKFLOW.md",
        "tools/architecture-sync/scan_architecture.py",
        ".agents/skills/sync-architecture-docs/SKILL.md",
        ".claude/skills/sync-architecture-docs/SKILL.md",
    )
    guard_documents: dict[str, Any] = {}
    for item in document_preconditions:
        document_text = read_text(root, item["document"], warnings) or ""
        guard_documents[item["document"]] = {
            "before_sha256": item["before_sha256"],
            "header_hash": document_header_hash(document_text),
            "headings": [
                heading
                for heading in documents[item["document"]]["headings"]
                if heading["level"] <= 3
            ],
            "protected_sections": protected_section_hashes(
                document_text, item["sections"]
            ),
            "sections": item["sections"],
        }
    guard_payload = {
        "allowed_documents": [item["document"] for item in update_plan],
        "content_fingerprint": content_fingerprint,
        "current_commit": head,
        "documents": guard_documents,
        "evidence_sample": [
            {
                "exists": bool(
                    (safe_candidate(root, path) or Path("__missing__")).exists()
                ),
                "path": path,
            }
            for path in evidence_allowlist[:24]
        ],
        "guard_version": 1,
        "initial_worktree": worktree_guard(root),
        "plan_id": plan_id,
        "protected_control_files": {
            path: file_fingerprint(root, path, warnings) for path in control_paths
        },
        "scope": scope,
    }
    verification_guard = encode_guard(guard_payload)
    return {
        "analysis_ready": True,
        **snapshot_core,
        "confirmation": {
            "accepted_phrase": "确认更新项目文档",
            "plan_id": plan_id,
            "required": requires_update,
            "scanner_write_capability": False,
            "short_phrase": "确认",
            "short_phrase_requires_same_pending_context": True,
        },
        "impact_mapping_version": 1,
        "impacts": impacts,
        "mode": mode,
        "next_action": "await_confirmation" if requires_update else "none",
        "non_architecture_changes": non_architecture,
        "plan_id": plan_id,
        "planned_updates": update_plan,
        "requires_update": requires_update,
        "schema_version": SCHEMA_VERSION,
        "snapshot_id": snapshot_id,
        "update_context": {
            "document_preconditions": document_preconditions,
            "evidence_allowlist": evidence_allowlist,
            "git_actions_allowed": False,
            "patch_method": "apply_patch",
            "plan_id": plan_id,
            "read_policy": "minimum_needed_from_allowlist",
            "sync_metadata": {
                "checked_date_format": "YYYY-MM-DD",
                "commit": head,
                "has_uncommitted_evidence": bool(working_tree_evidence),
                "marker": (
                    f"<!-- architecture-sync-commit: {head} -->" if head else None
                ),
            },
            "verification_guard": verification_guard,
            "working_tree_evidence": working_tree_evidence,
        },
        "warnings": warnings_list,
        "write_policy": {
            "current_plan_documents": list(targets),
            "never_modify": [
                "core/ARCHITECTURE.md",
                "business_code",
                "configuration",
                "migrations",
                "tests",
                "deployment_files",
                "git_index",
                "git_commits",
                "git_remote",
            ],
            "write_authorized": False,
        },
    }


def verify_snapshot(root: Path, token: str, scope: str) -> dict[str, Any]:
    errors: list[dict[str, Any]] = []
    warnings: list[str] = []
    checked_code_paths: set[str] = set()
    guard = decode_guard(token)
    plan_id = str(guard.get("plan_id", ""))
    guard_scope = guard.get("scope")
    if guard_scope != scope:
        errors.append(
            verification_error(
                "scope_mismatch",
                "校验范围与已确认计划不一致。",
                "使用原计划相同的 --scope 重新执行 verify。",
            )
        )
    allowed_documents = set(guard.get("allowed_documents", []))
    if not allowed_documents or not allowed_documents.issubset(set(TARGET_DOCUMENTS)):
        errors.append(
            verification_error(
                "allowed_documents_invalid",
                "校验令牌没有有效的目标文档白名单。",
                "丢弃当前令牌并重新运行 plan。",
            )
        )

    head = current_head(root)
    if head != guard.get("current_commit"):
        errors.append(
            verification_error(
                "head_commit_changed",
                "计划生成后 HEAD 已发生变化。",
                "停止写入并重新运行 plan 获取新确认。",
            )
        )

    fingerprint_warnings: set[str] = set()
    current_fingerprint, fingerprinted_count = scope_content_fingerprint(
        root, str(guard_scope), fingerprint_warnings
    )
    if current_fingerprint != guard.get("content_fingerprint"):
        errors.append(
            verification_error(
                "code_state_changed",
                "计划生成后相关代码事实发生变化。",
                "停止当前更新，重新扫描并重新确认。",
            )
        )
    warnings.extend(sorted(fingerprint_warnings))

    for path, expected in guard.get("protected_control_files", {}).items():
        current = file_fingerprint(root, path, set())
        if current != expected:
            errors.append(
                verification_error(
                    "control_file_changed",
                    "同步工作流、扫描器或 Skill 在执行期间发生变化。",
                    "停止验证，恢复控制文件或重新生成计划。",
                    path=path,
                )
            )

    errors.extend(
        compare_worktree_guards(
            guard.get("initial_worktree", {}), worktree_guard(root), allowed_documents
        )
    )

    document_results = []
    for document in sorted(allowed_documents):
        guard_document = guard.get("documents", {}).get(document)
        if not isinstance(guard_document, dict):
            errors.append(
                verification_error(
                    "document_guard_missing",
                    "目标文档缺少修改前保护信息。",
                    "丢弃当前令牌并重新运行 plan。",
                    document=document,
                )
            )
            continue
        read_warnings: set[str] = set()
        data = read_bytes(root, document, read_warnings)
        if data is None:
            errors.append(
                verification_error(
                    "target_document_missing",
                    "目标架构文档不存在或无法读取。",
                    "恢复目标文档后重新生成计划。",
                    document=document,
                )
            )
            continue
        warnings.extend(sorted(read_warnings))
        try:
            text = data.decode("utf-8-sig")
        except UnicodeDecodeError:
            errors.append(
                verification_error(
                    "target_document_not_utf8",
                    "目标架构文档不是有效 UTF-8。",
                    "将文档恢复为 UTF-8 编码后重新验证。",
                    document=document,
                )
            )
            continue
        current_sha = hashlib.sha256(data).hexdigest()
        changed = current_sha != guard_document.get("before_sha256")
        if not changed:
            warnings.append(f"planned_document_unchanged:{document}")
        current_headings = headings_from_text(text)
        errors.extend(
            validate_headings(document, guard_document.get("headings", []), current_headings)
        )
        if document_header_hash(text) != guard_document.get("header_hash"):
            errors.append(
                verification_error(
                    "document_header_changed",
                    "文档头部除核对日期和同步标记外发生了计划外变化。",
                    "恢复文档标题和说明，只保留允许更新的日期与同步标记。",
                    document=document,
                )
            )
        current_sections = {record["key"]: record["hash"] for record in section_records(text)}
        changed_protected_sections = [
            key
            for key, expected_hash in guard_document.get("protected_sections", {}).items()
            if current_sections.get(key) != expected_hash
        ]
        if changed_protected_sections:
            errors.append(
                verification_error(
                    "unplanned_section_changed",
                    "计划外章节正文发生变化。",
                    "恢复计划外章节，只保留 planned_updates 指定范围内的局部补丁。",
                    document=document,
                    sections=changed_protected_sections,
                )
            )
        errors.extend(validate_relative_links(root, document, text))
        errors.extend(validate_document_safety(document, text))
        reference_errors, checked = validate_code_references(root, document, text)
        errors.extend(reference_errors)
        checked_code_paths.update(checked)

        if changed:
            markers = [value.lower() for value in SYNC_MARKER_RE.findall(text)]
            if len(markers) != 1 or markers[0] != guard.get("current_commit"):
                errors.append(
                    verification_error(
                        "sync_marker_invalid",
                        "已修改文档缺少唯一且正确的同步 commit 标记。",
                        "在文首保留一条指向计划 HEAD 的 architecture-sync-commit 标记。",
                        document=document,
                    )
                )
            if not re.search(r"(?m)^>\s*最后核对：\d{4}-\d{2}-\d{2}。\s*$", text):
                errors.append(
                    verification_error(
                        "checked_date_missing",
                        "已修改文档缺少格式正确的最后核对日期。",
                        "在文首使用“> 最后核对：YYYY-MM-DD。”格式更新日期。",
                        document=document,
                    )
                )
        document_results.append(
            {
                "changed": changed,
                "document": document,
                "headings_checked": len(current_headings),
            }
        )

    evidence_checked = []
    for item in guard.get("evidence_sample", []):
        path = item.get("path", "")
        if not path or is_forbidden(path):
            errors.append(
                verification_error(
                    "evidence_path_invalid",
                    "校验令牌包含越界或敏感证据路径。",
                    "丢弃当前令牌并重新运行 plan。",
                )
            )
            continue
        current_exists = bool((safe_candidate(root, path) or Path("__missing__")).exists())
        if current_exists != bool(item.get("exists")):
            errors.append(
                verification_error(
                    "evidence_presence_changed",
                    "计划证据文件的存在状态发生变化。",
                    "停止当前更新，重新扫描并重新确认。",
                    path=path,
                )
            )
        elif current_exists:
            evidence_checked.append(path)

    status = "passed" if not errors else "failed"
    return {
        "mode": "verify",
        "plan_id": plan_id,
        "schema_version": SCHEMA_VERSION,
        "verification": {
            "checks": {
                "allowed_documents": sorted(allowed_documents),
                "code_references_checked": len(checked_code_paths),
                "evidence_files_checked": len(evidence_checked),
                "fingerprinted_files_checked": fingerprinted_count,
                "head_commit": head,
                "scope": guard_scope,
            },
            "documents": document_results,
            "errors": errors,
            "status": status,
            "warnings": sorted(set(warnings)),
        },
    }


def parse_arguments(arguments: Sequence[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Read-only AirShelf architecture scanner")
    parser.add_argument("mode", choices=("plan", "verify"))
    parser.add_argument(
        "--scope",
        choices=("all", "backend", "frontend"),
        default="all",
        help="Limit the current plan to both documents or one architecture domain",
    )
    parser.add_argument(
        "--guard",
        help="Verification guard emitted by plan mode; required by verify mode",
    )
    parser.add_argument(
        "--repo-root",
        type=Path,
        default=Path.cwd(),
        help="Any path inside the Git repository (default: current directory)",
    )
    return parser.parse_args(arguments)


def main(arguments: Sequence[str] | None = None) -> int:
    args = parse_arguments(arguments)
    try:
        root = locate_repository(args.repo_root)
        if args.mode == "verify":
            if not args.guard:
                payload = {
                    "mode": "verify",
                    "schema_version": SCHEMA_VERSION,
                    "verification": {
                        "errors": [
                            verification_error(
                                "verification_guard_required",
                                "verify 模式缺少 plan 生成的校验令牌。",
                                "先运行 plan，确认后传入 update_context.verification_guard。",
                            )
                        ],
                        "status": "failed",
                        "warnings": [],
                    },
                }
                print(json.dumps(payload, ensure_ascii=True, sort_keys=True, indent=2))
                return 3
            payload = verify_snapshot(root, args.guard, args.scope)
        else:
            payload = build_snapshot(root, args.mode, args.scope)
    except ScanError as error:
        if args.mode == "verify":
            payload = {
                "mode": "verify",
                "schema_version": SCHEMA_VERSION,
                "verification": {
                    "errors": [
                        verification_error(
                            str(error),
                            "校验令牌无效或只读校验无法继续。",
                            "丢弃当前令牌，重新运行 plan 并重新确认。",
                        )
                    ],
                    "status": "failed",
                    "warnings": [],
                },
            }
        else:
            payload = {
                "error": str(error),
                "mode": args.mode,
                "schema_version": SCHEMA_VERSION,
            }
        print(json.dumps(payload, ensure_ascii=True, sort_keys=True, indent=2))
        return 2
    print(json.dumps(payload, ensure_ascii=True, sort_keys=True, indent=2))
    if args.mode == "verify" and payload["verification"]["status"] != "passed":
        return 3
    return 0


if __name__ == "__main__":
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8", errors="backslashreplace")
    raise SystemExit(main())