ping-mee-self-service-platform/_bmad/scripts/setup.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# ///
"""Set up or report on the project BMad runtime from the installed bmod.toml files."""

from __future__ import annotations

import argparse
import copy
import datetime
import json
import os
import re
import shutil
import sys
import tomllib
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path, PurePosixPath
from typing import NamedTuple

sys.dont_write_bytecode = True

MANIFEST_NAME = "bmod.toml"
RETIRED_NAME = "retired.toml"
QUESTION_KEYS = frozenset({"key", "prompt", "default"})
OPTIONAL_QUESTION_KEYS = frozenset({"scope"})
QUESTION_SCOPES = ("team", "user")
UPDATE_SOURCE_PREFIXES = ("github:", "https://", "file:", "plugin:")
MODULE_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]*\Z")
SKILL_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]*\Z")
RESERVED_MODULE_DIRS = frozenset({"_config", "custom", "modules", "scripts"})
TEAM_CONFIG = "_bmad/config.toml"
USER_CONFIG = "_bmad/custom/config.user.toml"
CUSTOM_GITIGNORE = "*.user.toml\n"
GITIGNORE_COVERS_USER_CONFIG = frozenset({"*.user.toml", "config.user.toml", "*.toml", "*"})

SEMVER = re.compile(
    r"(?P<major>0|[1-9][0-9]*)\."
    r"(?P<minor>0|[1-9][0-9]*)\."
    r"(?P<patch>0|[1-9][0-9]*)"
    r"(?:-(?P<prerelease>"
    r"(?:0|[1-9][0-9]*|[0-9A-Za-z-]*[A-Za-z-][0-9A-Za-z-]*)"
    r"(?:\.(?:0|[1-9][0-9]*|[0-9A-Za-z-]*[A-Za-z-][0-9A-Za-z-]*))*"
    r"))?"
    r"(?:\+(?P<build>[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?\Z"
)
SOURCE_READ_LIMIT = 1024 * 1024
_MISSING = object()


class Requirement(NamedTuple):
    skill: str
    version: str | None
    source: str | None


class KnowledgeEntry(NamedTuple):
    path: PurePosixPath
    skills: tuple[str, ...] | None


class Rename(NamedTuple):
    old: str
    new: str


class ConfigQuestion(NamedTuple):
    module: str
    key: str
    prompt: str
    default: str
    scope: str = "team"


class ParsedBmod(NamedTuple):
    code: str
    version: str
    update_source: str
    skills: tuple[str, ...] | None
    knowledge: tuple[KnowledgeEntry, ...]
    questions: tuple[ConfigQuestion, ...]
    required_skills: tuple[Requirement, ...]
    recommended_skills: tuple[Requirement, ...]
    pre_install_message: str = ""
    post_install_message: str = ""


class ParsedRetired(NamedTuple):
    renamed: tuple[Rename, ...] = ()
    removed: tuple[str, ...] = ()


class ParsedSkill(NamedTuple):
    bmod: str | None
    source: str | None
    scripts: tuple[PurePosixPath, ...]
    required_skills: tuple[Requirement, ...]
    recommended_skills: tuple[Requirement, ...]


class ParsedFile(NamedTuple):
    bmod: ParsedBmod | None
    skill: ParsedSkill | None


class InstalledFile(NamedTuple):
    folder: str
    source: Path
    file: Path
    parsed: ParsedFile


class InstalledModule(NamedTuple):
    module: str
    folder: str
    source: Path
    file: Path
    parsed: ParsedBmod
    skills: tuple[str, ...]
    absent_skills: tuple[str, ...]
    members: tuple[InstalledFile, ...]
    questions: tuple[ConfigQuestion, ...]
    retired: ParsedRetired = ParsedRetired()


class Retired(NamedTuple):
    name: str
    module: str
    renamed_to: str | None
    update_source: str
    record_folder: Path


class Retirement(NamedTuple):
    in_use: tuple[dict[str, object], ...]
    renames: tuple[tuple[str, str], ...]
    unmoved: tuple[tuple[str, str], ...]
    unused: tuple[tuple[str, str], ...]
    install_offers: tuple[dict[str, object], ...]


class Installation(NamedTuple):
    files: tuple[InstalledFile, ...]
    modules: tuple[InstalledModule, ...]
    missing_records: tuple[dict[str, object], ...]
    problems: tuple[dict[str, object], ...]
    roots: tuple[Path, ...] = ()
    folders: dict[str, Path] = {}
    duplicates: tuple[dict[str, object], ...] = ()


class PlainTree(NamedTuple):
    directories: tuple[PurePosixPath, ...]
    files: tuple[tuple[PurePosixPath, bytes], ...]


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Set up and repair {project-root}/_bmad, or report on the installed BMad modules."
    )
    parser.add_argument("--project-root", type=Path, required=True)
    parser.add_argument("--skill", type=Path, required=True)
    parser.add_argument("--module", help="limit the run to one module, by code or by bmod-<code>")
    parser.add_argument(
        "--root",
        type=Path,
        action="append",
        default=[],
        help="an active skills folder, repeated; the first holding a skill wins",
    )
    parser.add_argument("--module-answers", type=Path)
    parser.add_argument(
        "--list-config-questions",
        action="store_true",
        help="print unanswered installed-module questions as JSON",
    )
    parser.add_argument(
        "--status",
        action="store_true",
        help="report on the installation without changing files",
    )
    parser.add_argument(
        "--remove-retired",
        nargs="+",
        metavar="SKILL",
        help="delete these renamed or removed skills from the project's skills folders",
    )
    parser.add_argument(
        "--remove-copies",
        nargs="+",
        metavar="PATH",
        help="delete these copies of duplicated skills, as duplicate_skills lists them",
    )
    parser.add_argument(
        "--source-record",
        nargs=2,
        metavar=("SOURCE", "FOLDER"),
        help="print the version and pre-install message of the module record FOLDER at SOURCE",
    )
    args = parser.parse_args(argv)
    project_root = args.project_root.resolve()
    skill_root = args.skill.resolve()
    roots = tuple(root.resolve() for root in args.root)
    if args.source_record is not None:
        if (
            args.status
            or args.list_config_questions
            or args.module_answers is not None
            or args.remove_retired is not None
            or args.remove_copies is not None
        ):
            parser.error("--source-record cannot be combined with other modes")
        # setup.md adds --module to every call when the user names a module; it does not apply here.
        print_json(source_record_report(project_root, *args.source_record))
        return 0
    if args.remove_retired is not None or args.remove_copies is not None:
        if (
            args.status
            or args.list_config_questions
            or args.module_answers is not None
            or (args.remove_retired is not None and args.remove_copies is not None)
        ):
            parser.error("--remove-retired and --remove-copies cannot be combined with other modes")
        if args.remove_copies is not None:
            print_json(remove_copies(project_root, skill_root, args.remove_copies, roots=roots))
        else:
            print_json(remove_retired(project_root, skill_root, args.remove_retired, module=args.module, roots=roots))
        return 0
    if args.status:
        if args.list_config_questions or args.module_answers is not None:
            parser.error("--status cannot be combined with questions or answers")
        print_json(status_report(project_root, skill_root, module=args.module, roots=roots))
        return 0
    if args.list_config_questions:
        if args.module_answers is not None:
            parser.error("--list-config-questions cannot be combined with answer files")
        listing = list_config_questions(project_root, skill_root, module=args.module, roots=roots)
        print_json(listing)
        return 0
    report = setup(
        project_root,
        skill_root,
        module=args.module,
        roots=roots,
        module_answers=(load_module_answers(args.module_answers) if args.module_answers is not None else None),
        module_answers_source=args.module_answers,
    )
    print_json(report)
    return 0


def print_json(value: object) -> None:
    print(json.dumps(value, ensure_ascii=False, default=str))


def setup(
    project_root: Path,
    skill_root: Path,
    *,
    module: str | None = None,
    module_answers: dict[tuple[str, str], str] | None = None,
    module_answers_source: Path | None = None,
    roots: tuple[Path, ...] = (),
) -> dict[str, object]:
    """Create what is missing, repair what is stale, add new answers, and report what was done."""
    bmad = project_root / "_bmad"
    reject_unusable_bmad(project_root)
    scripts_src, config_src = payload(skill_root)
    installation = discover_installation(skill_root, roots)
    selected, unknown = select_module(installation, module, mode="setup")
    if unknown is not None:
        return unknown
    scoped = installation.modules if selected is None else (selected,)

    existing_text, merged, base_text = team_config_plan(project_root, config_src)
    user_existing_text, user_existing = existing_user_config(project_root)
    user_merged = copy.deepcopy(user_existing)

    pending = find_pending_questions(scoped, merged, user_existing, project_root)
    answers = validate_module_answers(module_answers, pending, source=module_answers_source)
    team_added: list[tuple[tuple[str, ...], str]] = []
    user_added: list[tuple[tuple[str, ...], str]] = []
    for question in pending:
        path = ("modules", question.module, *question.key.split("."))
        value = answers[(question.module, question.key)]
        if question.scope == "user":
            set_missing_value(user_merged, path, value, user_config_path(project_root))
            user_added.append((path, value))
        else:
            set_missing_value(merged, path, value, project_root / "_bmad" / "config.toml")
            team_added.append((path, value))
    # base_text is the file's own text only when the template adds nothing to it.
    team_text = base_text if base_text == existing_text else None
    config_text = text_with_answers(team_text, team_added, merged) if team_added else base_text
    user_text = text_with_answers(user_existing_text, user_added, user_merged) if user_added else None
    if user_added:
        reject_unwritable_user_config(project_root)

    retirement = retirement_report(project_root, skill_root, installation, retired_skills(scoped, installation.modules))

    module_trees: dict[str, PlainTree] = {}
    for installed in scoped:
        reject_unusable_module_root(bmad / installed.module)
        module_trees[installed.module] = declared_scripts_tree(read_module_scripts(installed))

    created = not bmad.exists()
    done = {"missing": "created", "stale": "repaired", "current": "current"}
    shared_state = done[tree_state(bmad / "scripts", read_plain_tree(scripts_src))]
    module_states = {code: done[tree_state(bmad / code / "scripts", tree)] for code, tree in module_trees.items()}
    config_state = team_config_state(project_root, existing_text, config_text)
    gitignore_state = custom_gitignore_state(project_root)
    custom = bmad / "custom"
    changed = (
        created
        or shared_state != "current"
        or config_state != "current"
        or bool(user_added)
        or gitignore_state == "missing"
        or not (custom.exists() or custom.is_symlink())
        or any(state != "current" for state in module_states.values())
        or bool(retirement.renames)
    )
    if changed:
        materialize_bmad(
            project_root,
            scripts_src,
            config_text,
            module_trees,
            user_config_text=user_text,
            custom_renames=retirement.renames,
        )
    output = project_root / output_folder(config_text)
    if not output.exists() and not output.is_symlink():
        changed = True
    ensure_dir(output)

    unmet = unmet_requirements(installation, skill_root, module=None if selected is None else selected.module)
    recommended = unmet_recommendations(installation, skill_root, module=None if selected is None else selected.module)
    remaining = find_pending_questions(installation.modules, merged, user_merged, project_root)
    next_command = next_step(installation.missing_records, unmet, setup_owed=bool(remaining))
    problems = [*installation.problems, *custom_gitignore_problems(gitignore_state)]
    return {
        "mode": "setup",
        "status": "created" if created else "repaired" if changed else "current",
        "changed": changed,
        "module": None if selected is None else selected.module,
        "bmad": bmad_report(installation, skill_root),
        "shared_scripts": shared_state,
        "config": config_state,
        "custom_gitignore": "created" if gitignore_state == "missing" else gitignore_state,
        "modules": [
            {
                **module_summary(installed, project_root),
                "scripts": module_states[installed.module],
                **message_json("post_install_message", installed.parsed.post_install_message),
            }
            for installed in scoped
        ],
        "answers_added": [
            {
                "module": question.module,
                "key": question.key,
                "scope": question.scope,
                "file": scope_file(question.scope),
            }
            for question in pending
        ],
        "answers": current_answers(scoped, merged, user_merged),
        "pending_questions": [question_json(question) for question in remaining],
        "unmet_requirements": unmet,
        "unmet_recommendations": recommended,
        "missing_module_records": list(installation.missing_records),
        "duplicate_skills": duplicates_json(installation.duplicates, project_root),
        "problems": problems,
        **retirement_json(retirement),
        "current": (next_command is None and not unmet and not problems and not installation.missing_records),
        "next": next_command,
    }


def payload(skill_root: Path) -> tuple[Path, Path]:
    scripts_src = skill_root / "scripts"
    assets_src = skill_root / "assets"
    config_src = assets_src / "config.template.toml"
    resolve_config = scripts_src / "resolve_config.py"
    for directory in (scripts_src, assets_src):
        if not directory.is_dir():
            raise Exception(f"missing directory: {directory}")
    for file in (resolve_config, config_src):
        if not file.is_file():
            raise Exception(f"missing file: {file}")
    return (scripts_src, config_src)


def team_config_plan(project_root: Path, config_src: Path) -> tuple[str | None, dict, str]:
    """The team file's text, its values with the template's new keys filled in, and the text setup would write."""
    template_text = fill_team_config(config_src.read_text(encoding="utf-8"), project_root)
    template = parse_toml(template_text, config_src)
    existing_text, existing = existing_team_config(project_root)
    merged = fill_keep(template, existing)
    if not isinstance(merged, dict):
        raise Exception(f"invalid team config: {project_root / '_bmad' / 'config.toml'}")
    base_text = existing_text if existing_text is not None and merged == existing else render_toml(merged)
    return existing_text, merged, base_text


def team_config_state(project_root: Path, existing_text: str | None, config_text: str) -> str:
    if existing_text is None:
        return "created"
    path = project_root / "_bmad" / "config.toml"
    if path.is_symlink() or fill_toml(existing_text, config_text) != existing_text:
        return "updated"
    return "current"


def reject_unusable_module_root(module_root: Path) -> None:
    if module_root.is_symlink() or (module_root.exists() and not module_root.is_dir()):
        raise Exception(f"module runtime is not a plain directory: {module_root}")


def pending_config_questions(
    project_root: Path,
    skill_root: Path,
    modules: tuple[InstalledModule, ...],
) -> tuple[ConfigQuestion, ...]:
    _scripts, config_src = payload(skill_root)
    _existing_text, merged, _base_text = team_config_plan(project_root, config_src)
    _user_text, user_config = existing_user_config(project_root)
    return find_pending_questions(modules, merged, user_config, project_root)


def list_config_questions(
    project_root: Path, skill_root: Path, *, module: str | None = None, roots: tuple[Path, ...] = ()
) -> object:
    """The pending questions as a JSON list, or the unknown-module report."""
    reject_unusable_bmad(project_root)
    installation = discover_installation(skill_root, roots)
    selected, unknown = select_module(installation, module, mode="list-config-questions")
    if unknown is not None:
        return unknown
    scoped = installation.modules if selected is None else (selected,)
    return [question_json(question) for question in pending_config_questions(project_root, skill_root, scoped)]


def question_json(question: ConfigQuestion) -> dict[str, str]:
    return {
        "module": question.module,
        "key": question.key,
        "prompt": question.prompt,
        "default": question.default,
        "scope": question.scope,
    }


def scope_file(scope: str) -> str:
    return USER_CONFIG if scope == "user" else TEAM_CONFIG


def current_answers(
    modules: tuple[InstalledModule, ...],
    team: dict,
    user: dict,
) -> dict[str, list[dict[str, object]]]:
    answers: dict[str, list[dict[str, object]]] = {}
    for installed in modules:
        entries: list[dict[str, object]] = []
        for question in installed.questions:
            config = user if question.scope == "user" else team
            value = lookup(config, ("modules", question.module, *question.key.split(".")))
            if value is _MISSING:
                continue
            entries.append(
                {
                    "key": question.key,
                    "scope": question.scope,
                    "file": scope_file(question.scope),
                    "value": value,
                }
            )
        if entries:
            answers[installed.module] = entries
    return answers


def lookup(data: object, keys: tuple[str, ...]) -> object:
    current = data
    for key in keys:
        if not isinstance(current, dict) or key not in current:
            return _MISSING
        current = current[key]
    return current


def reject_unusable_bmad(project_root: Path) -> None:
    bmad = project_root / "_bmad"
    if bmad.is_symlink():
        target = bmad.resolve()
        raise Exception(
            f"{bmad} is a symlink to {target}; setup replaces "
            f"_bmad in place, so run it with --project-root "
            f"{target.parent} to fix the real installation"
        )
    if bmad.exists() and not bmad.is_dir():
        raise Exception(f"existing BMad runtime is not a directory: {bmad}")


def user_config_path(project_root: Path) -> Path:
    return project_root / "_bmad" / "custom" / "config.user.toml"


def existing_user_config(project_root: Path) -> tuple[str | None, dict]:
    path = user_config_path(project_root)
    if not path.exists() and not path.is_symlink():
        return None, {}
    if not path.is_file():
        raise Exception(f"user config is not a file: {path}")
    try:
        text = path.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as error:
        raise Exception(f"cannot read user config {path}: {error}") from error
    return text, parse_toml(text, path)


def reject_unwritable_user_config(project_root: Path) -> None:
    """Setup writes a user answer only into a plain file in a plain folder."""
    path = user_config_path(project_root)
    for candidate in (path.parent, path):
        if candidate.is_symlink():
            raise Exception(f"cannot add a user answer: {candidate} is a symlink")
    if path.parent.exists() and not path.parent.is_dir():
        raise Exception(f"cannot add a user answer: {path.parent} is not a directory")


def custom_gitignore_state(project_root: Path) -> str:
    custom = project_root / "_bmad" / "custom"
    if custom.is_symlink() or (custom.exists() and not custom.is_dir()):
        return "skipped"
    gitignore = custom / ".gitignore"
    if not gitignore.exists() and not gitignore.is_symlink():
        return "missing"
    try:
        lines = {line.strip() for line in gitignore.read_text(encoding="utf-8").splitlines()}
    except (OSError, UnicodeError):
        return "unprotected"
    return "current" if lines & GITIGNORE_COVERS_USER_CONFIG else "unprotected"


def custom_gitignore_problems(state: str) -> list[dict[str, object]]:
    if state != "unprotected":
        return []
    return [
        {
            "kind": "custom-gitignore",
            "message": (
                f"_bmad/custom/.gitignore has no line that ignores {USER_CONFIG}, so user answers may be "
                "committed; add the line *.user.toml"
            ),
        }
    ]


def retired_skills(
    scoped: tuple[InstalledModule, ...],
    installed: tuple[InstalledModule, ...],
) -> tuple[Retired, ...]:
    """Old names the modules renamed or removed. A name an installed module still lists is not retired."""
    current = {name for module in installed for name in (*module.skills, *module.absent_skills)}
    retired: list[Retired] = []
    for module in scoped:
        entries = [(rename.old, rename.new) for rename in module.retired.renamed]
        entries += [(name, None) for name in module.retired.removed]
        retired.extend(
            Retired(name, module.module, new, module.parsed.update_source, module.source)
            for name, new in entries
            if name not in current
        )
    return tuple(retired)


def skill_folders(project_root: Path, installation: Installation) -> tuple[Path, ...]:
    """Where retired skills are looked for: the active roots, which may be global, and every `.<tool>/skills`
    in the project, where a classic installer may have left copies for tools this host does not load.
    """
    candidates: list[Path] = list(installation.roots)
    try:
        entries = sorted(project_root.iterdir(), key=lambda path: path.name)
    except OSError:
        entries = []
    inside = project_root.resolve()
    active = {root.resolve() for root in installation.roots}
    # A tool folder linked outside the project is not a project copy; deleting through it would hit the target.
    candidates += [
        folder
        for entry in entries
        if entry.name.startswith(".") and (folder := entry / "skills").is_dir()
        if folder.resolve().is_relative_to(inside) or folder.resolve() in active
    ]
    folders: list[Path] = []
    seen: set[Path] = set()
    for folder in candidates:
        resolved = folder.resolve()
        if resolved not in seen:
            seen.add(resolved)
            folders.append(folder)
    return tuple(folders)


def is_global(path: Path, project_root: Path) -> bool:
    """A skill or skills folder outside the project belongs to the skills CLI's global scope."""
    return not path.is_relative_to(project_root)


def shown_path(path: Path, project_root: Path) -> str:
    """Relative inside the project, `~/`-based under the home folder, absolute otherwise."""
    if path.is_relative_to(project_root):
        return path.relative_to(project_root).as_posix()
    home = Path.home()
    if path.is_relative_to(home):
        return "~/" + path.relative_to(home).as_posix()
    return path.as_posix()


def lock_path(project_root: Path, path: Path) -> Path:
    """The skills CLI lock of the scope a skill folder is in: the project's, or the global one."""
    if not is_global(path, project_root):
        return project_root / "skills-lock.json"
    state = os.environ.get("XDG_STATE_HOME")
    if state:
        return Path(state) / "skills" / ".skill-lock.json"
    return Path.home() / ".agents" / ".skill-lock.json"


def present(path: Path) -> bool:
    return path.exists() or path.is_symlink()


def retirement_report(
    project_root: Path, skill_root: Path, installation: Installation, retired: tuple[Retired, ...]
) -> Retirement:
    """Where the retired skills are still installed or customized, and what setup does about it."""
    folders = skill_folders(project_root, installation)
    custom = project_root / "_bmad" / "custom"
    plain_custom = custom.is_dir() and not custom.is_symlink()
    in_use: list[dict[str, object]] = []
    renames: list[tuple[str, str]] = []
    unmoved: list[tuple[str, str]] = []
    unused: list[tuple[str, str]] = []
    used: set[str] = set()
    for skill in retired:
        paths = [folder / skill.name for folder in folders if present(folder / skill.name)]
        if paths:
            used.add(skill.name)
            in_use.append(
                {
                    "skill": skill.name,
                    "module": skill.module,
                    "renamed_to": skill.renamed_to,
                    "paths": [shown_path(path, project_root) for path in paths],
                    "global": any(is_global(path, project_root) for path in paths),
                }
            )
        if not plain_custom:
            continue
        for suffix in (".toml", ".user.toml"):
            old = f"{skill.name}{suffix}"
            if not present(custom / old):
                continue
            used.add(skill.name)
            if skill.renamed_to is None:
                unused.append((skill.name, old))
                continue
            new = f"{skill.renamed_to}{suffix}"
            taken = present(custom / new) or any(target == new for _old, target in renames)
            (unmoved if taken else renames).append((old, new))
    offers: dict[str, dict[str, object]] = {}
    for skill in retired:
        new = skill.renamed_to
        if new is None or skill.name not in used or new in offers or new in installation.folders:
            continue
        offers[new] = {
            "skill": new,
            "replaces": skill.name,
            "module": skill.module,
            "install": install_command(
                skill.update_source, new, global_install=is_global(skill.record_folder, project_root)
            ),
        }
    return Retirement(tuple(in_use), tuple(renames), tuple(unmoved), tuple(unused), tuple(offers.values()))


def retirement_json(retirement: Retirement) -> dict[str, object]:
    def custom(name: str) -> str:
        return f"_bmad/custom/{name}"

    return {
        "retired_skills": list(retirement.in_use),
        "custom_renames": [{"from": custom(old), "to": custom(new)} for old, new in retirement.renames],
        "custom_not_renamed": [{"from": custom(old), "to": custom(new)} for old, new in retirement.unmoved],
        "custom_unused": [{"skill": skill, "file": custom(name)} for skill, name in retirement.unused],
        "install_offers": list(retirement.install_offers),
    }


def remove_retired(
    project_root: Path,
    skill_root: Path,
    names: list[str],
    *,
    module: str | None = None,
    roots: tuple[Path, ...] = (),
) -> dict[str, object]:
    """Delete retired skills from the skills folders and drop them from the skills CLI locks."""
    installation = discover_installation(skill_root, roots)
    selected, unknown = select_module(installation, module, mode="remove-retired")
    if unknown is not None:
        return unknown
    scoped = installation.modules if selected is None else (selected,)
    retired = {skill.name for skill in retired_skills(scoped, installation.modules)}
    names = list(dict.fromkeys(names))
    for name in names:
        if name not in retired:
            raise Exception(f"{name!r} is not a renamed or removed skill of an installed module")
    folders = skill_folders(project_root, installation)
    targets = [(name, folder / name) for folder in folders for name in names if present(folder / name)]
    return {"mode": "remove-retired", **remove_skill_paths(project_root, folders, targets)}


def remove_copies(
    project_root: Path,
    skill_root: Path,
    paths: list[str],
    *,
    roots: tuple[Path, ...] = (),
) -> dict[str, object]:
    """Delete copies of duplicated skills, keeping at least one copy of each."""
    installation = discover_installation(skill_root, roots)
    copies: dict[str, tuple[str, Path]] = {}
    listed: list[tuple[str, list[str]]] = []
    for duplicate in installation.duplicates:
        skill = str(duplicate["skill"])
        shown = []
        for folder in duplicate["folders"]:  # type: ignore[attr-defined]
            copies[shown_path(folder, project_root)] = (skill, folder)
            shown.append(shown_path(folder, project_root))
        listed.append((skill, shown))
    paths = list(dict.fromkeys(paths))
    for path in paths:
        if path not in copies:
            raise Exception(f"{path!r} is not a copy listed in duplicate_skills")
    for skill, shown in listed:
        if all(path in paths for path in shown):
            raise Exception(f"removing every copy of {skill!r} is not a duplicate cleanup")
    targets = [copies[path] for path in paths]
    return {
        "mode": "remove-copies",
        **remove_skill_paths(project_root, skill_folders(project_root, installation), targets),
    }


def remove_skill_paths(
    project_root: Path,
    folders: tuple[Path, ...],
    targets: list[tuple[str, Path]],
) -> dict[str, object]:
    """Delete skill folders, then drop each name from the lock of its scope once no copy is left there.

    Every affected lock is read before anything is deleted, so a bad lock stops the run first.
    """
    locks: dict[Path, dict | None] = {}
    for _name, path in targets:
        lock_file = lock_path(project_root, path)
        if lock_file not in locks:
            locks[lock_file] = read_skills_lock(lock_file)
    removed: list[str] = []
    for _name, path in targets:
        if path.is_symlink() or path.is_file():
            path.unlink()
        elif path.is_dir():
            shutil.rmtree(path)
        else:
            continue
        removed.append(shown_path(path, project_root))
    changed: list[dict[str, object]] = []
    for lock_file, lock in locks.items():
        if lock is None:
            continue
        scope_folders = [folder for folder in folders if lock_path(project_root, folder) == lock_file]
        dropped = [
            name
            for name in dict.fromkeys(name for name, path in targets if lock_path(project_root, path) == lock_file)
            if name in lock["skills"] and not any(present(folder / name) for folder in scope_folders)
        ]
        if not dropped:
            continue
        for name in dropped:
            del lock["skills"][name]
        ending = "\n" if lock_file.read_text(encoding="utf-8").endswith("\n") else ""
        lock_file.write_text(json.dumps(lock, indent=2, ensure_ascii=False) + ending, encoding="utf-8")
        changed.append({"file": shown_path(lock_file, project_root), "entries_removed": dropped})
    return {"removed": removed, "locks": changed}


def read_skills_lock(path: Path) -> dict | None:
    """The skills CLI's lock, read before anything is deleted so a bad file stops the run."""
    if not path.is_file():
        return None
    try:
        lock = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError) as error:
        raise Exception(f"cannot read {path}: {error}") from error
    if not isinstance(lock, dict) or not isinstance(lock.get("skills"), dict):
        raise Exception(f"{path} has no skills table")
    return lock


def select_module(
    installation: Installation,
    name: str | None,
    *,
    mode: str,
) -> tuple[InstalledModule | None, dict[str, object] | None]:
    """The module a name means, or the report that lists what is installed."""
    if name is None:
        return None, None
    for names in (
        lambda installed: installed.module,
        lambda installed: f"bmod-{installed.module}",
        lambda installed: installed.folder,
    ):
        for installed in installation.modules:
            if name == names(installed):
                return installed, None
    wanted = {name, f"bmod-{name}"}
    return None, {
        "mode": mode,
        "status": "unknown-module",
        "changed": False,
        "module": name,
        "installed_modules": [installed.module for installed in installation.modules],
        "missing_module_records": [record for record in installation.missing_records if record["bmod"] in wanted],
    }


def module_summary(installed: InstalledModule, project_root: Path) -> dict[str, object]:
    global_install = is_global(installed.source, project_root)
    return {
        "module": installed.module,
        "folder": installed.folder,
        "version": installed.parsed.version,
        "update_source": installed.parsed.update_source,
        "skills": list(installed.skills),
        "scope": "global" if global_install else "project",
        "absent_skills": list(installed.absent_skills),
        "absent_install": install_command(
            installed.parsed.update_source, *installed.absent_skills, global_install=global_install
        ),
    }


def bmad_report(installation: Installation, skill_root: Path) -> dict[str, object]:
    """The bmad skill in use. Its version is its module's, and unknown when that record is absent."""
    report: dict[str, object] = {"skill": skill_root.name, "version": None, "module": None}
    resolved = skill_root.resolve()
    by_folder = {installed.folder: installed for installed in installation.files}
    for installed in installation.files:
        if installed.source.resolve() != resolved:
            continue
        record = installed.parsed.bmod
        if record is None and installed.parsed.skill is not None and installed.parsed.skill.bmod is not None:
            other = by_folder.get(installed.parsed.skill.bmod)
            record = other.parsed.bmod if other is not None else None
        if record is not None:
            report.update({"version": record.version, "module": record.code})
        break
    return report


def unmet_requirements(
    installation: Installation,
    skill_root: Path,
    *,
    module: str | None = None,
) -> list[dict[str, object]]:
    """Required skills the current install does not satisfy."""
    return unmet_entries(installation, skill_root, "required_skills", module)


def unmet_recommendations(
    installation: Installation,
    skill_root: Path,
    *,
    module: str | None = None,
) -> list[dict[str, object]]:
    """Recommended skills that are absent or too old. Worth offering, never a fault."""
    return unmet_entries(installation, skill_root, "recommended_skills", module)


def unmet_entries(
    installation: Installation,
    skill_root: Path,
    field: str,
    module: str | None,
) -> list[dict[str, object]]:
    """A module's list is checked once for the module, a skill's own list once for the skill."""
    by_folder = {installed.folder: installed for installed in installation.files}
    declared: list[tuple[str, str | None, str, Requirement]] = []
    for installed in installation.modules:
        for requirement in getattr(installed.parsed, field):
            declared.append((installed.folder, installed.module, installed.parsed.update_source, requirement))
    for installed in installation.files:
        skill = installed.parsed.skill
        if skill is None:
            continue
        record = installed.parsed.bmod
        if record is None and skill.bmod is not None and skill.bmod in by_folder:
            record = by_folder[skill.bmod].parsed.bmod
        default_source = skill.source if skill.source is not None else record.update_source if record else None
        for requirement in getattr(skill, field):
            declared.append((installed.folder, record.code if record else None, default_source or "", requirement))

    unmet: list[dict[str, object]] = []
    for folder, code, default_source, requirement in declared:
        if requirement.skill == folder or (module is not None and code != module):
            continue
        state, present = requirement_check(installation.folders, requirement)
        if state is None:
            continue
        source = requirement.source if requirement.source is not None else default_source
        unmet.append(
            {
                "skill": folder,
                "module": code,
                "requires": requirement.skill,
                "minimum": requirement.version,
                "installed": present,
                "state": state,
                "source": source,
                "channel": requirement_channel(source),
                "install": fix_command(state, source, requirement.skill),
            }
        )
    return unmet


def skill_path(skills: Path | dict[str, Path], name: str) -> Path | None:
    """An installed skill's folder: from one skills folder, or from the folders found across the roots."""
    if isinstance(skills, dict):
        return skills.get(name)
    path = skills / name
    return path if path.is_dir() else None


def requirement_check(skills: Path | dict[str, Path], requirement: Requirement) -> tuple[str | None, str | None]:
    """Why a requirement is unmet, with the version found. The state is None when it is met."""
    if skill_path(skills, requirement.skill) is None:
        return "missing", None
    if requirement.version is None:
        return None, None
    installed = skill_module_version(skills, requirement.skill)
    if installed is None:
        if names_a_module_record(skills, requirement.skill):
            # Its record is absent, which is reported as a missing module record.
            return None, None
        # A copy from before module records has no version to read, and that
        # copy is what a minimum version exists to catch.
        return "unknown-version", None
    return requirement_state(installed, requirement.version), installed


def names_a_module_record(skills: Path | dict[str, Path], skill: str) -> bool:
    folder = skill_path(skills, skill)
    try:
        parsed = read_bmod_file(folder / MANIFEST_NAME) if folder is not None else None
    except Exception:
        return False
    return parsed is not None and parsed.skill is not None and parsed.skill.bmod is not None


def skill_module_version(skills: Path | dict[str, Path], skill: str) -> str | None:
    """The version of the module an installed skill belongs to.

    Skills carry no version. None means the skill has no bmod.toml or its
    module record is absent, so there is nothing to compare.
    """
    try:
        folder = skill_path(skills, skill)
        parsed = read_bmod_file(folder / MANIFEST_NAME) if folder is not None else None
        if parsed is None:
            return None
        if parsed.bmod is not None:
            return parsed.bmod.version
        if parsed.skill is None or parsed.skill.bmod is None:
            return None
        record_folder = skill_path(skills, parsed.skill.bmod)
        if record_folder is None:
            return None
        record = read_bmod_file(record_folder / MANIFEST_NAME)
    except Exception:
        return None
    if record is None or record.bmod is None:
        return None
    return record.bmod.version


def read_bmod_file(path: Path) -> ParsedFile | None:
    if not path.is_file():
        return None
    try:
        raw = path.read_bytes()
    except OSError as error:
        raise Exception(f"cannot read bmod file {path}: {error}") from error
    return parse_bmod_file(path, raw)


def requirement_state(installed: str | None, minimum: str) -> str | None:
    """Why an installed version fails a minimum, or None when it meets it.

    The development branch carries `X-next` until `X` is released, and that
    build already holds everything `X` will. SemVer orders it below `X`, which
    would report every skill on a development install as outdated.
    """
    if installed is None:
        return "missing"
    comparison = compare_semver(installed, minimum)
    if comparison is None:
        return "unorderable"
    if comparison >= 0:
        return None
    have = parse_orderable_semver(installed)
    want = parse_orderable_semver(minimum)
    if have is not None and want is not None and have[0] == want[0] and want[1] is None and have[1] == ("next",):
        return None
    return "outdated"


def requirement_channel(source: str) -> str:
    """How a missing or stale requirement is installed, so help offers the right command."""
    if source.startswith("plugin:"):
        return "plugin"
    if source.startswith("file:"):
        return "local"
    return "skills-cli"


def install_command(source: str, *skills: str, global_install: bool = False) -> str | None:
    """The `npx skills` command that installs these skills, or None when the source has no such command."""
    if not skills or not source.startswith("github:"):
        return None
    owner, repository, *_tree = source.removeprefix("github:").split("/")
    return f"npx skills add {owner}/{repository} --skill {' '.join(skills)}" + (" -g" if global_install else "")


UPDATE_FIXES = ("outdated", "unknown-version")


def fix_command(state: str, source: str, skill: str) -> str | None:
    """Adding a skill does not raise its module's version, so an outdated one is updated instead."""
    if state == "missing":
        return install_command(source, skill)
    if state in UPDATE_FIXES and requirement_channel(source) == "skills-cli":
        return "npx skills update"
    return None


def next_step(
    missing_records: tuple[dict[str, object], ...],
    unmet: list[dict[str, object]],
    *,
    update_available: bool = False,
    setup_owed: bool = False,
    module: str | None = None,
) -> str | None:
    """The one command to run next: install what is absent, then update, then setup."""
    for record in missing_records:
        if record["install"] is not None:
            return str(record["install"])
    for entry in unmet:
        if entry["state"] == "missing" and entry["install"] is not None:
            return str(entry["install"])
    outdated = any(entry["state"] in UPDATE_FIXES and entry["channel"] == "skills-cli" for entry in unmet)
    if outdated or update_available:
        return "npx skills update"
    if setup_owed:
        return "bmad setup" if module is None else f"bmad setup {module}"
    return None


def existing_team_config(project_root: Path) -> tuple[str | None, dict]:
    path = project_root / "_bmad" / "config.toml"
    if not path.exists() and not path.is_symlink():
        return None, {}
    if not path.is_file():
        raise Exception(f"team config is not a file: {path}")
    try:
        text = path.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as error:
        raise Exception(f"cannot read team config {path}: {error}") from error
    return text, parse_toml(text, path)


def parse_toml(text: str, source: Path | str) -> dict:
    try:
        return tomllib.loads(text)
    except tomllib.TOMLDecodeError as error:
        raise Exception(f"cannot parse TOML {source}: {error}") from error


def discover_installation(skill_root: Path, roots: tuple[Path, ...] = ()) -> Installation:
    """Every bmod.toml in the active roots, sorted into module records, their skills, and problems.

    The folder the bmad skill runs from is always a root. The first root holding a skill wins it, so a
    project copy shadows a global one; the other copies are reported as duplicates.
    """
    problems: list[dict[str, object]] = []
    roots = unique_folders((*roots, skill_root.parent))
    folders, duplicates = locate_skills(roots, skill_root.parent)
    files = discover_installed_files(folders, problems)
    by_folder = {installed.folder: installed for installed in files}
    winners = select_module_records(files, problems)

    modules: list[InstalledModule] = []
    for code in sorted(winners):
        record_file = winners[code]
        record = record_file.parsed.bmod
        assert record is not None
        listed = member_names(record_file)
        present = tuple(name for name in listed if name in folders)
        members: list[InstalledFile] = []
        for name in present:
            member = by_folder.get(name)
            if member is None or member.parsed.skill is None:
                continue
            if member is record_file or (member.parsed.bmod is None and member.parsed.skill.bmod == record_file.folder):
                members.append(member)
                continue
            detail = (
                f"names {member.parsed.skill.bmod!r} as its bmod"
                if member.parsed.bmod is None
                else "is a module record of its own"
            )
            problems.append(
                {
                    "kind": "membership",
                    "skill": name,
                    "bmod": record_file.folder,
                    "message": f"{record_file.file} lists the skill {name!r}, but {member.file} {detail}",
                }
            )
        try:
            retired = read_retired_file(record_file.source)
        except Exception as error:
            problems.append({"kind": "retired-file", "folder": record_file.folder, "message": str(error)})
            retired = ParsedRetired()
        modules.append(
            InstalledModule(
                code,
                record_file.folder,
                record_file.source,
                record_file.file,
                record,
                present,
                tuple(name for name in listed if name not in present),
                tuple(members),
                record.questions,
                retired,
            )
        )

    missing_records: list[dict[str, object]] = []
    for installed in files:
        skill = installed.parsed.skill
        if skill is not None and installed.parsed.bmod is not None and installed.folder not in member_names(installed):
            problems.append(
                {
                    "kind": "membership",
                    "skill": installed.folder,
                    "bmod": installed.folder,
                    "message": f"{installed.file} holds [bmod] and [skill], but its skills list leaves out {installed.folder!r}",
                }
            )
        if skill is None or installed.parsed.bmod is not None or skill.bmod is None:
            continue
        record_file = by_folder.get(skill.bmod)
        if record_file is None or record_file.parsed.bmod is None:
            source = skill.source or ""
            missing_records.append(
                {
                    "skill": installed.folder,
                    "bmod": skill.bmod,
                    "source": source,
                    "channel": requirement_channel(source),
                    "install": install_command(source, skill.bmod),
                }
            )
        elif installed.folder not in member_names(record_file):
            problems.append(
                {
                    "kind": "membership",
                    "skill": installed.folder,
                    "bmod": skill.bmod,
                    "message": (
                        f"{installed.file} names {skill.bmod!r} as its bmod, but "
                        f"{record_file.file} does not list the skill {installed.folder!r}"
                    ),
                }
            )
    return Installation(files, tuple(modules), tuple(missing_records), tuple(problems), roots, folders, duplicates)


def unique_folders(folders: tuple[Path, ...]) -> tuple[Path, ...]:
    """The folders in order, each once, however it is reached."""
    unique: list[Path] = []
    seen: set[Path] = set()
    for folder in folders:
        resolved = folder.resolve()
        if resolved not in seen:
            seen.add(resolved)
            unique.append(folder)
    return tuple(unique)


def locate_skills(roots: tuple[Path, ...], own_folder: Path) -> tuple[dict[str, Path], tuple[dict[str, object], ...]]:
    """Each skill's folder in the first root that holds it, and the BMad skills installed in more than one.

    The bmad skill's own folder must be readable; an unreadable extra root is skipped.
    """
    copies: dict[str, list[Path]] = {}
    for root in roots:
        try:
            entries = sorted(root.iterdir(), key=lambda path: path.name)
        except OSError as error:
            if root.resolve() == own_folder.resolve():
                raise Exception(f"cannot inspect installed skills {root}: {error}") from error
            continue
        for entry in entries:
            if entry.is_dir():
                copies.setdefault(entry.name, []).append(entry)
    folders = {name: paths[0] for name, paths in copies.items()}
    duplicates: list[dict[str, object]] = []
    for name, paths in sorted(copies.items()):
        distinct = unique_folders(tuple(paths))
        if len(distinct) > 1 and any((path / MANIFEST_NAME).is_file() for path in distinct):
            duplicates.append({"skill": name, "folders": distinct})
    return folders, tuple(duplicates)


def copy_version(folder: Path) -> str | None:
    """The module version of one installed copy, read through the record beside it."""
    try:
        parsed = read_bmod_file(folder / MANIFEST_NAME)
        if parsed is None:
            return None
        if parsed.bmod is not None:
            return parsed.bmod.version
        if parsed.skill is None or parsed.skill.bmod is None:
            return None
        record = read_bmod_file(folder.parent / parsed.skill.bmod / MANIFEST_NAME)
    except Exception:
        return None
    return record.bmod.version if record is not None and record.bmod is not None else None


def duplicates_json(duplicates: tuple[dict[str, object], ...], project_root: Path) -> list[dict[str, object]]:
    """Each duplicate with the copy in use first, and whether a copy not in use is newer."""
    report: list[dict[str, object]] = []
    for duplicate in duplicates:
        folders: tuple[Path, ...] = duplicate["folders"]  # type: ignore[assignment]
        copies = [
            {
                "path": shown_path(folder, project_root),
                "global": is_global(folder, project_root),
                "version": copy_version(folder),
            }
            for folder in folders
        ]
        used = copies[0]["version"]
        newer_elsewhere = any(
            (compare_semver(str(copy["version"]), str(used)) or 0) > 0
            for copy in copies[1:]
            if copy["version"] is not None and used is not None
        )
        report.append(
            {
                "skill": duplicate["skill"],
                "used": copies[0]["path"],
                "newer_copy_unused": newer_elsewhere,
                "copies": copies,
            }
        )
    return report


def discover_installed_files(folders: dict[str, Path], problems: list[dict[str, object]]) -> tuple[InstalledFile, ...]:
    """One unusable file must not stop the install: it becomes a problem and its folder is skipped."""
    files: list[InstalledFile] = []
    for name in sorted(folders):
        sibling = folders[name]
        path = sibling / MANIFEST_NAME
        try:
            parsed = read_bmod_file(path)
        except Exception as error:
            problems.append({"kind": "bmod-file", "folder": sibling.name, "message": str(error)})
            continue
        if parsed is not None:
            files.append(InstalledFile(sibling.name, sibling, path, parsed))
    return tuple(files)


def select_module_records(
    files: tuple[InstalledFile, ...],
    problems: list[dict[str, object]],
) -> dict[str, InstalledFile]:
    """One record per module code. Files arrive sorted by folder name, so the first one wins."""
    casefolded: dict[str, InstalledFile] = {}
    winners: dict[str, InstalledFile] = {}
    for installed in files:
        record = installed.parsed.bmod
        if record is None:
            continue
        previous = casefolded.get(record.code.casefold())
        if previous is None:
            casefolded[record.code.casefold()] = installed
            winners[record.code] = installed
            continue
        assert previous.parsed.bmod is not None
        if previous.parsed.bmod.code != record.code:
            raise Exception(
                "installed module codes differ only by case: "
                f"{previous.parsed.bmod.code!r} from {previous.file} and "
                f"{record.code!r} from {installed.file}"
            )
        problems.append(
            {
                "kind": "duplicate-module",
                "module": record.code,
                "folder": installed.folder,
                "kept": previous.folder,
                "message": (
                    f"module code {record.code!r} is declared by {previous.file} and by "
                    f"{installed.file}; the first is used"
                ),
            }
        )
    return winners


def member_names(record_file: InstalledFile) -> tuple[str, ...]:
    record = record_file.parsed.bmod
    assert record is not None
    if record.skills is not None:
        return record.skills
    return (record_file.folder,) if record_file.parsed.skill is not None else ()


def read_module_scripts(installed: InstalledModule) -> tuple[tuple[PurePosixPath, bytes], ...]:
    """The scripts a module's installed skills place in `_bmad/<code>/scripts/`."""
    placed: dict[PurePosixPath, tuple[bytes, Path]] = {}
    scripts: list[tuple[PurePosixPath, bytes]] = []
    for member in installed.members:
        assert member.parsed.skill is not None
        for relative in member.parsed.skill.scripts:
            content = read_declared_script(member.source, relative, member.file)
            destination = PurePosixPath(*relative.parts[1:])
            previous = placed.get(destination)
            if previous is None:
                placed[destination] = (content, member.file)
                scripts.append((relative, content))
            elif previous[0] != content:
                raise Exception(
                    f"module {installed.module!r} has two different scripts for "
                    f"{destination.as_posix()!r}: {previous[1]} and {member.file}"
                )
    return tuple(scripts)


def parse_bmod_file(path: Path, raw: bytes) -> ParsedFile:
    """Read the fields BMad uses and ignore every other key and table.

    An author may add keys of their own, and a newer file may carry keys this
    version predates. Neither may stop a skill from installing.
    """
    try:
        source = raw.decode("utf-8")
    except UnicodeError as error:
        raise Exception(f"invalid bmod file {path}: {error}") from error
    data = parse_toml(source, path)
    bmod_table = data.get("bmod")
    skill_table = data.get("skill")
    if bmod_table is None and skill_table is None:
        raise Exception(f"bmod file {path} must hold a [bmod] table, a [skill] table, or both")
    for name, table in (("bmod", bmod_table), ("skill", skill_table)):
        if table is not None and not isinstance(table, dict):
            raise Exception(f"bmod file {path} field {name!r} must be a table")
    bmod = parse_bmod_table(bmod_table, path) if bmod_table is not None else None
    skill = parse_skill_table(skill_table, path, standalone=bmod is None) if skill_table is not None else None
    return ParsedFile(bmod, skill)


def parse_bmod_table(table: dict, path: Path) -> ParsedBmod:
    code = required_string(table, "bmod", "code", path)
    if MODULE_NAME.fullmatch(code) is None or code.casefold() in RESERVED_MODULE_DIRS:
        raise Exception(f"bmod file {path} field 'bmod.code' has unsafe value {code!r}")
    version = required_string(table, "bmod", "version", path)
    update_source = required_string(table, "bmod", "update_source", path)
    validate_source(update_source, "bmod.update_source", path)
    skills = table.get("skills")
    return ParsedBmod(
        code,
        version,
        update_source,
        parse_skill_names(skills, "bmod.skills", path) if skills is not None else None,
        parse_knowledge(table.get("knowledge"), path),
        parse_questions(table.get("config_questions"), code, path),
        parse_requirements(table.get("required_skills"), "bmod.required_skills", path),
        parse_requirements(table.get("recommended_skills"), "bmod.recommended_skills", path),
        optional_string(table, "bmod", "pre_install_message", path),
        optional_string(table, "bmod", "post_install_message", path),
    )


def read_retired_file(folder: Path) -> ParsedRetired:
    """The skills a module renamed or removed, from the `retired.toml` beside its record."""
    path = folder / RETIRED_NAME
    if not path.is_file():
        return ParsedRetired()
    try:
        data = parse_toml(path.read_bytes().decode("utf-8"), path)
    except (OSError, UnicodeError) as error:
        raise Exception(f"cannot read {path}: {error}") from error
    renamed = parse_renamed(data.get("renamed"), path)
    removed = parse_skill_names(data.get("removed", []), "removed", path)
    retired = [rename.old for rename in renamed] + list(removed)
    repeated = next((name for name in retired if retired.count(name) > 1), None)
    if repeated is not None:
        raise Exception(f"bmod file {path} retires {repeated!r} more than once in renamed and removed")
    return ParsedRetired(renamed, removed)


def parse_renamed(value: object, path: Path) -> tuple[Rename, ...]:
    """Skills the module renamed, as `{ from, to }` tables."""
    if value is None:
        return ()
    if not isinstance(value, list):
        raise Exception(f"bmod file {path} field 'renamed' must be a list of tables")
    renamed: list[Rename] = []
    for index, entry in enumerate(value):
        field = f"renamed[{index}]"
        if not isinstance(entry, dict):
            raise Exception(f"bmod file {path} field {field} must be a table")
        names: list[str] = []
        for key in ("from", "to"):
            name = entry.get(key)
            if not isinstance(name, str) or SKILL_NAME.fullmatch(name) is None:
                raise Exception(f"bmod file {path} field '{field}.{key}' must be a skill name; found {name!r}")
            names.append(name)
        if names[0] == names[1]:
            raise Exception(f"bmod file {path} field {field} renames {names[0]!r} to itself")
        renamed.append(Rename(*names))
    return tuple(renamed)


def parse_skill_table(table: dict, path: Path, *, standalone: bool) -> ParsedSkill:
    bmod: str | None = None
    source: str | None = None
    if standalone:
        bmod = required_string(table, "skill", "bmod", path)
        if SKILL_NAME.fullmatch(bmod) is None:
            raise Exception(f"bmod file {path} field 'skill.bmod' has unsafe value {bmod!r}")
        source = required_string(table, "skill", "source", path)
        validate_source(source, "skill.source", path)
    return ParsedSkill(
        bmod,
        source,
        parse_scripts(table.get("scripts"), path),
        parse_requirements(table.get("required_skills"), "skill.required_skills", path),
        parse_requirements(table.get("recommended_skills"), "skill.recommended_skills", path),
    )


def validate_source(value: str, field: str, path: Path) -> None:
    prefix = next(
        (candidate for candidate in UPDATE_SOURCE_PREFIXES if value.startswith(candidate)),
        None,
    )
    if prefix is None or not value.removeprefix(prefix):
        raise Exception(f"bmod file {path} field {field!r} must name a source")
    if prefix == "github:":
        github_parts = value.removeprefix(prefix).split("/")
        if len(github_parts) < 2 or any(not part for part in github_parts):
            raise Exception(f"bmod file {path} field {field!r} github source must name owner/repo")
    if prefix == "https://" and any(character.isspace() for character in value):
        raise Exception(f"bmod file {path} field {field!r} must be a valid HTTPS URL")


def parse_skill_names(value: object, field: str, path: Path) -> tuple[str, ...]:
    if not isinstance(value, list):
        raise Exception(f"bmod file {path} field {field!r} must be a list of skill names")
    names: list[str] = []
    for entry in value:
        if not isinstance(entry, str) or SKILL_NAME.fullmatch(entry) is None:
            raise Exception(f"bmod file {path} field {field!r} has unsafe skill name {entry!r}")
        if entry in names:
            raise Exception(f"bmod file {path} field {field!r} repeats {entry!r}")
        names.append(entry)
    return tuple(names)


def parse_path(entry: object, field: str, path: Path, seen: list[PurePosixPath]) -> PurePosixPath:
    if not isinstance(entry, str) or not entry:
        raise Exception(f"bmod file {path} field {field!r} has invalid value {entry!r}")
    relative = safe_skill_relative(entry)
    if relative is None:
        raise Exception(f"bmod file {path} field {field!r} has unsafe value {entry!r}")
    if relative in seen:
        raise Exception(f"bmod file {path} field {field!r} repeats {entry!r}")
    return relative


def parse_knowledge(value: object, path: Path) -> tuple[KnowledgeEntry, ...]:
    """The module's help documents, as paths inside the bmod folder, each with the skills it covers."""
    if value is None:
        return ()
    if not isinstance(value, list):
        raise Exception(f"bmod file {path} field 'bmod.knowledge' must be a list of tables")
    knowledge: list[KnowledgeEntry] = []
    for index, entry in enumerate(value):
        field = f"bmod.knowledge[{index}]"
        if not isinstance(entry, dict):
            raise Exception(f"bmod file {path} field {field} must be a table")
        relative = parse_path(entry.get("path"), f"{field}.path", path, [item.path for item in knowledge])
        skills = entry.get("skills", "*")
        if skills == "*":
            knowledge.append(KnowledgeEntry(relative, None))
            continue
        if isinstance(skills, str):
            raise Exception(f"bmod file {path} field '{field}.skills' must be \"*\" or a list of skill names")
        knowledge.append(KnowledgeEntry(relative, parse_skill_names(skills, f"{field}.skills", path)))
    return tuple(knowledge)


def safe_skill_relative(entry: str) -> PurePosixPath | None:
    """A bmod.toml path that cannot escape the skill folder, or None if it can.

    Shared with validate_manifests.py and knowledge.py so one rule decides
    this everywhere. A URL parses as an ordinary relative path and a Windows
    drive prefix makes a later join discard the skill folder, so both are
    refused by name. pathlib drops "." components itself, so only ".." and an
    empty final component need checking.
    """
    if not entry or "://" in entry or "\\" in entry or ":" in entry:
        return None
    relative = PurePosixPath(entry)
    if relative.is_absolute() or ".." in relative.parts or not relative.name:
        return None
    return relative


def parse_requirements(value: object, field: str, path: Path) -> tuple[Requirement, ...]:
    """A flat list of skills. A plain name comes from the declaring file's own source."""
    if value is None:
        return ()
    if not isinstance(value, list):
        raise Exception(f"bmod file {path} field {field!r} must be a list of skills")
    requirements: list[Requirement] = []
    for index, entry in enumerate(value):
        item = f"{field}[{index}]"
        if isinstance(entry, str):
            requirement = Requirement(entry, None, None)
        elif isinstance(entry, dict):
            skill = entry.get("skill")
            if not isinstance(skill, str):
                raise Exception(f"bmod file {path} field '{item}.skill' must be a string and is required")
            source = entry.get("source")
            if not isinstance(source, str):
                raise Exception(f"bmod file {path} field '{item}.source' must be a string and is required")
            validate_source(source, f"{item}.source", path)
            minimum = entry.get("version")
            if minimum is not None:
                if not isinstance(minimum, str):
                    raise Exception(f"bmod file {path} field '{item}.version' must be a string")
                if parse_orderable_semver(minimum) is None:
                    raise Exception(
                        f"bmod file {path} field '{item}.version' must be an orderable version; found {minimum!r}"
                    )
            requirement = Requirement(skill, minimum, source)
        else:
            raise Exception(f"bmod file {path} field {item!r} must be a skill name or a table")
        if SKILL_NAME.fullmatch(requirement.skill) is None:
            raise Exception(f"bmod file {path} field {item!r} has unsafe skill name {requirement.skill!r}")
        if any(requirement.skill == other.skill for other in requirements):
            raise Exception(f"bmod file {path} field {field!r} repeats {requirement.skill!r}")
        requirements.append(requirement)
    return tuple(requirements)


def required_string(table: dict, name: str, field: str, path: Path) -> str:
    value = table.get(field)
    if not isinstance(value, str) or not value.strip():
        raise Exception(f"bmod file {path} field '{name}.{field}' must be a non-empty string")
    return value


def optional_string(table: dict, name: str, field: str, path: Path | str) -> str:
    value = table.get(field, "")
    if not isinstance(value, str):
        raise Exception(f"bmod file {path} field '{name}.{field}' must be a string")
    return value


def parse_questions(value: object, module: str, path: Path) -> tuple[ConfigQuestion, ...]:
    if value is None:
        return ()
    if not isinstance(value, list):
        raise Exception(f"bmod file {path} field 'bmod.config_questions' must be a list")
    questions: list[ConfigQuestion] = []
    seen: list[str] = []
    for index, question in enumerate(value):
        field = f"bmod.config_questions[{index}]"
        if not isinstance(question, dict):
            raise Exception(f"bmod file {path} field {field} must be a mapping")
        keys = set(question)
        if not QUESTION_KEYS <= keys or not keys <= QUESTION_KEYS | OPTIONAL_QUESTION_KEYS:
            missing = sorted(QUESTION_KEYS - keys)
            unknown = sorted(keys - QUESTION_KEYS - OPTIONAL_QUESTION_KEYS, key=str)
            detail = f"missing key {missing[0]!r}" if missing else f"unknown key {unknown[0]!r}"
            raise Exception(f"bmod file {path} field {field} has {detail}")
        for key in QUESTION_KEYS:
            if not isinstance(question[key], str):
                raise Exception(f"bmod file {path} field {field}.{key} must be a string")
        scope = question.get("scope", "team")
        if scope not in QUESTION_SCOPES:
            raise Exception(f'bmod file {path} field {field}.scope must be "team" or "user"; found {scope!r}')
        prompt = question["prompt"]
        key = question["key"]
        if not prompt.strip():
            raise Exception(f"bmod file {path} field {field}.prompt must be non-empty")
        if not key or any(not part or part != part.strip() for part in key.split(".")):
            raise Exception(f"bmod file {path} field {field}.key must be a non-empty dotted key")
        if key == module or key.startswith(f"{module}."):
            raise Exception(f"bmod file {path} field {field}.key {key!r} must not start with module {module!r}")
        conflict = conflicting_question_key(seen, key)
        if conflict is not None:
            raise Exception(f"bmod file {path} config question key {key!r} conflicts with {conflict!r}")
        seen.append(key)
        questions.append(ConfigQuestion(module, key, prompt, question["default"], scope))
    return tuple(questions)


def conflicting_question_key(keys: list[str], candidate: str) -> str | None:
    for key in keys:
        if key == candidate or key.startswith(f"{candidate}.") or candidate.startswith(f"{key}."):
            return key
    return None


def parse_scripts(value: object, path: Path) -> tuple[PurePosixPath, ...]:
    if value is None:
        return ()
    if not isinstance(value, list):
        raise Exception(f"bmod file {path} field 'skill.scripts' must be a list")
    scripts: list[PurePosixPath] = []
    for entry in value:
        if not isinstance(entry, str) or not entry:
            raise Exception(f"bmod file {path} field 'skill.scripts' has invalid value {entry!r}")
        relative = PurePosixPath(entry)
        if (
            relative.is_absolute()
            or "\\" in entry
            or len(relative.parts) < 2
            or relative.parts[0] != "scripts"
            or ".." in relative.parts
            or "." in relative.parts
        ):
            raise Exception(f"bmod file {path} field 'skill.scripts' has unsafe value {entry!r}")
        scripts.append(relative)
    return tuple(scripts)


def read_declared_script(skill_root: Path, relative: PurePosixPath, file: Path) -> bytes:
    root = skill_root.resolve()
    candidate = root.joinpath(*relative.parts)
    try:
        resolved = candidate.resolve(strict=True)
        resolved.relative_to(root)
    except (OSError, RuntimeError, ValueError) as error:
        raise Exception(f"bmod file {file} declares unsafe or missing script {relative.as_posix()!r}") from error
    if not resolved.is_file():
        raise Exception(f"bmod file {file} declared script {relative.as_posix()!r} is not a file")
    try:
        return resolved.read_bytes()
    except OSError as error:
        raise Exception(f"cannot read script {resolved} declared by {file}: {error}") from error


def status_report(
    project_root: Path, skill_root: Path, *, module: str | None = None, roots: tuple[Path, ...] = ()
) -> dict[str, object]:
    """Everything `bmad status` says. Reads only; nothing under the project is written."""
    reject_unusable_bmad(project_root)
    bmad = project_root / "_bmad"
    installation = discover_installation(skill_root, roots)
    selected, unknown = select_module(installation, module, mode="status")
    if unknown is not None:
        return unknown
    scoped = installation.modules if selected is None else (selected,)
    code = None if selected is None else selected.module
    problems = list(installation.problems)

    modules: list[dict[str, object]] = []
    update_states: list[str] = []
    for installed in scoped:
        update = module_update_report(project_root, installed)
        update_states.append(str(update["state"]))
        try:
            reject_unusable_module_root(bmad / installed.module)
            scripts = tree_state(
                bmad / installed.module / "scripts", declared_scripts_tree(read_module_scripts(installed))
            )
        except Exception as error:
            scripts = "could-not-check"
            problems.append({"kind": "scripts", "module": installed.module, "message": str(error)})
        modules.append(
            {
                **module_summary(installed, project_root),
                "scripts": scripts,
                "update": update,
            }
        )

    scripts_src, config_src = payload(skill_root)
    shared_state = tree_state(bmad / "scripts", read_plain_tree(scripts_src))
    existing_text, _merged, base_text = team_config_plan(project_root, config_src)
    output = project_root / output_folder(base_text)
    custom = bmad / "custom"
    pending = pending_config_questions(project_root, skill_root, scoped)
    unmet = unmet_requirements(installation, skill_root, module=code)
    gitignore_state = custom_gitignore_state(project_root)
    problems.extend(custom_gitignore_problems(gitignore_state))
    retirement = retirement_report(project_root, skill_root, installation, retired_skills(scoped, installation.modules))
    setup_owed = (
        not bmad.is_dir()
        or shared_state != "current"
        or team_config_state(project_root, existing_text, base_text) != "current"
        or not (custom.exists() or custom.is_symlink())
        or not (output.exists() or output.is_symlink())
        or bool(pending)
        or gitignore_state == "missing"
        or any(item["scripts"] in ("missing", "stale") for item in modules)
        or bool(retirement.renames)
    )
    next_command = next_step(
        installation.missing_records,
        unmet,
        update_available="newer-available" in update_states,
        setup_owed=setup_owed,
        module=code,
    )
    settled = ("current", "ahead", "plugin-managed", "could-not-check")
    return {
        "mode": "status",
        "module": code,
        "bmad_exists": bmad.is_dir(),
        "bmad": bmad_report(installation, skill_root),
        "shared_scripts": shared_state,
        "custom_gitignore": gitignore_state,
        "modules": modules,
        "missing_module_records": list(installation.missing_records),
        "duplicate_skills": duplicates_json(installation.duplicates, project_root),
        "pending_questions": [question_json(question) for question in pending],
        "unmet_requirements": unmet,
        "unmet_recommendations": unmet_recommendations(installation, skill_root, module=code),
        "problems": problems,
        **retirement_json(retirement),
        "current": (
            next_command is None
            and not unmet
            and not problems
            and not installation.missing_records
            and all(state in settled for state in update_states)
        ),
        "next": next_command,
    }


def module_update_report(project_root: Path, installed: InstalledModule) -> dict[str, object]:
    """How the installed module compares with the `[bmod] version` at its source."""
    update_source = installed.parsed.update_source
    if update_source.startswith("plugin:"):
        plugin = update_source.removeprefix("plugin:")
        return {
            "state": "plugin-managed",
            "plugin": plugin,
            "instruction": (
                f"this module ships inside the {plugin} plugin — update the "
                "plugin through its marketplace, not these files"
            ),
        }
    source = update_source
    try:
        source = source_file_location(project_root, update_source, installed.folder)
        source_version, pre_message = parse_source_record(source, read_source_file(source, update_source))
    except Exception as error:
        return {"state": "could-not-check", "source": source, "reason": str(error)}
    state = version_state(installed.parsed.version, source_version)
    return {
        "state": state,
        "source": source,
        "source_version": source_version,
        **(message_json("pre_install_message", pre_message) if state == "newer-available" else {}),
    }


def source_record_report(project_root: Path, update_source: str, folder: str) -> dict[str, object]:
    """The version and pre-install message of a module record at its source, before the module is added."""
    report: dict[str, object] = {"mode": "source-record", "folder": folder, "source": update_source}
    try:
        if SKILL_NAME.fullmatch(folder) is None:
            raise Exception(f"unsafe module record folder {folder!r}")
        validate_source(update_source, "source", Path("--source-record"))
        if update_source.startswith("plugin:"):
            raise Exception("a plugin source has no module record to read")
        source = source_file_location(project_root, update_source, folder)
        report["source"] = source
        version, pre_message = parse_source_record(source, read_source_file(source, update_source))
    except Exception as error:
        return {**report, "state": "could-not-check", "reason": str(error)}
    return {**report, "state": "read", "version": version, **message_json("pre_install_message", pre_message)}


def message_json(key: str, message: str) -> dict[str, str]:
    """An install message for a report; an empty one is left out, so it is never shown."""
    return {key: message} if message.strip() else {}


def source_file_location(project_root: Path, update_source: str, folder: str) -> str:
    quoted_folder = urllib.parse.quote(folder, safe="")
    quoted_name = urllib.parse.quote(MANIFEST_NAME, safe="")
    if update_source.startswith("file:"):
        root_text = update_source.removeprefix("file:")
        root = Path(root_text)
        if not root.is_absolute():
            root = project_root / root
        return str((root / folder / MANIFEST_NAME).resolve())
    if update_source.startswith("https://"):
        try:
            parsed = urllib.parse.urlsplit(update_source)
        except ValueError as error:
            raise Exception(f"invalid update_source {update_source!r}: {error}") from error
        return urllib.parse.urlunsplit(
            parsed._replace(path=(parsed.path.rstrip("/") + f"/{quoted_folder}/{quoted_name}"))
        )
    github = update_source.removeprefix("github:")
    owner, repository, *tree = github.split("/")
    # With no path the repo root is the skill, so its bmod.toml sits at the root.
    parts = (*tree, folder, MANIFEST_NAME) if tree else (MANIFEST_NAME,)
    path = "/".join(urllib.parse.quote(part, safe="") for part in parts)
    return (
        "https://raw.githubusercontent.com/"
        f"{urllib.parse.quote(owner, safe='')}/"
        f"{urllib.parse.quote(repository, safe='')}/main/{path}"
    )


def read_source_file(source: str, update_source: str) -> bytes:
    if update_source.startswith("file:"):
        path = Path(source)
        try:
            raw = path.read_bytes()
        except OSError as error:
            raise Exception(f"cannot read source bmod file {path}: {error}") from error
    else:
        request = urllib.request.Request(
            source,
            headers={"Accept": "text/plain", "User-Agent": "bmad-status"},
        )
        try:
            with urllib.request.urlopen(request, timeout=10) as response:
                raw = response.read(SOURCE_READ_LIMIT + 1)
        except (OSError, urllib.error.URLError) as error:
            raise Exception(f"cannot read source bmod file {source}: {error}") from error
    if len(raw) > SOURCE_READ_LIMIT:
        raise Exception(f"source bmod file {source} exceeds {SOURCE_READ_LIMIT} bytes")
    return raw


def parse_source_record(source: str, raw: bytes) -> tuple[str, str]:
    """The `[bmod]` version and pre-install message of a source record, from the one fetch status makes."""
    try:
        text = raw.decode("utf-8")
    except UnicodeError as error:
        raise Exception(f"invalid source bmod file {source}: {error}") from error
    table = parse_toml(text, source).get("bmod")
    version = table.get("version") if isinstance(table, dict) else None
    if not isinstance(version, str) or not version.strip():
        raise Exception(f"source bmod file {source} field 'bmod.version' must be a non-empty string")
    # Lenient: a bad message at the source must not hide a newer version.
    message = table.get("pre_install_message", "")
    return version, message if isinstance(message, str) else ""


def version_state(installed: str, source: str) -> str:
    comparison = compare_semver(installed, source)
    if comparison is None:
        return "differing-unordered"
    if comparison == 0:
        return "current"
    if comparison < 0:
        return "newer-available"
    return "ahead"


def compare_semver(left: str, right: str) -> int | None:
    left_parsed = parse_orderable_semver(left)
    right_parsed = parse_orderable_semver(right)
    if left_parsed is None or right_parsed is None:
        return None
    left_core, left_pre = left_parsed
    right_core, right_pre = right_parsed
    if left_core != right_core:
        return -1 if left_core < right_core else 1
    return compare_prerelease(left_pre, right_pre)


def parse_orderable_semver(
    value: str,
) -> tuple[tuple[int, int, int], tuple[str, ...] | None] | None:
    match = SEMVER.fullmatch(value)
    if match is None or "-dev" in value.casefold():
        return None
    prerelease = match.group("prerelease")
    return (
        (
            int(match.group("major")),
            int(match.group("minor")),
            int(match.group("patch")),
        ),
        tuple(prerelease.split(".")) if prerelease is not None else None,
    )


def compare_prerelease(left: tuple[str, ...] | None, right: tuple[str, ...] | None) -> int:
    if left is None or right is None:
        if left is right:
            return 0
        return 1 if left is None else -1
    for left_item, right_item in zip(left, right, strict=False):
        if left_item == right_item:
            continue
        left_numeric = left_item.isdigit()
        right_numeric = right_item.isdigit()
        if left_numeric and right_numeric:
            return -1 if int(left_item) < int(right_item) else 1
        if left_numeric != right_numeric:
            return -1 if left_numeric else 1
        return -1 if left_item < right_item else 1
    if len(left) == len(right):
        return 0
    return -1 if len(left) < len(right) else 1


def declared_scripts_tree(scripts: tuple[tuple[PurePosixPath, bytes], ...]) -> PlainTree:
    by_path = {PurePosixPath(*relative.parts[1:]): content for relative, content in scripts}
    if not by_path:
        # Git drops an empty directory, and a clone without it would report the module's scripts as missing.
        by_path = {PurePosixPath(".gitkeep"): b""}
    files = tuple(sorted(by_path.items(), key=lambda item: item[0].as_posix()))
    directories = {
        PurePosixPath(*relative.parts[:index])
        for relative, _content in files
        for index in range(1, len(relative.parts))
    }
    return PlainTree(tuple(sorted(directories, key=str)), files)


def read_plain_tree(root: Path) -> PlainTree:
    if not root.is_dir() or root.is_symlink():
        raise Exception(f"payload scripts are not a plain directory: {root}")
    directories: list[PurePosixPath] = []
    files: list[tuple[PurePosixPath, bytes]] = []
    try:
        entries = sorted(root.rglob("*"), key=lambda path: path.as_posix())
    except OSError as error:
        raise Exception(f"cannot inspect payload scripts {root}: {error}") from error
    for entry in entries:
        relative = PurePosixPath(entry.relative_to(root).as_posix())
        if entry.is_symlink():
            raise Exception(f"payload scripts contain a symlink: {entry}")
        if entry.is_dir():
            directories.append(relative)
            continue
        if not entry.is_file():
            raise Exception(f"payload scripts contain a non-file entry: {entry}")
        try:
            content = entry.read_bytes()
        except OSError as error:
            raise Exception(f"cannot read payload script {entry}: {error}") from error
        files.append((relative, content))
    return PlainTree(tuple(directories), tuple(files))


def tree_matches(root: Path, expected: PlainTree) -> bool:
    if not root.is_dir() or root.is_symlink():
        return False
    try:
        actual = read_plain_tree(root)
    except Exception:
        return False
    return actual == expected


def tree_state(root: Path, expected: PlainTree) -> str:
    if not root.exists() and not root.is_symlink():
        return "missing"
    return "current" if tree_matches(root, expected) else "stale"


def find_pending_questions(
    modules: tuple[InstalledModule, ...],
    config: dict,
    user_config: dict,
    project_root: Path,
) -> tuple[ConfigQuestion, ...]:
    """A team question is pending once per project, a user question once per person."""
    pending: list[ConfigQuestion] = []
    for installed in modules:
        for question in installed.questions:
            path = ("modules", question.module, *question.key.split("."))
            if question.scope == "user":
                answered = has_path(user_config, path, user_config_path(project_root))
            else:
                answered = has_path(config, path, project_root / "_bmad" / "config.toml")
            if not answered:
                pending.append(
                    question._replace(default=question.default.replace("{directory_name}", project_root.name))
                )
    return tuple(pending)


def has_path(data: object, path: tuple[str, ...], source: Path) -> bool:
    current = data
    for part in path:
        if not isinstance(current, dict):
            raise Exception(f"cannot inspect {'.'.join(path)}: parent value in {source} is not a table")
        if part not in current:
            return False
        current = current[part]
    return True


def load_module_answers(path: Path) -> dict[tuple[str, str], str]:
    if not path.is_file():
        raise Exception(f"missing file: {path}")
    try:
        data = tomllib.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, tomllib.TOMLDecodeError) as error:
        raise Exception(f"cannot parse module answers {path}: {error}") from error
    if not data:
        return {}
    if set(data) != {"modules"} or not isinstance(data["modules"], dict):
        raise Exception(f"--module-answers {path} must contain only module answer tables")
    answers: dict[tuple[str, str], str] = {}
    for module, values in data["modules"].items():
        if not isinstance(module, str) or not isinstance(values, dict):
            raise Exception(f"--module-answers {path} has an invalid module table")
        flatten_module_answers(path, module, values, (), answers)
    return answers


def flatten_module_answers(
    source: Path,
    module: str,
    values: dict,
    prefix: tuple[str, ...],
    answers: dict[tuple[str, str], str],
) -> None:
    for key, value in values.items():
        parts = (*prefix, str(key))
        if isinstance(value, dict):
            flatten_module_answers(source, module, value, parts, answers)
            continue
        dotted = ".".join(parts)
        if not isinstance(value, str):
            raise Exception(f"--module-answers {source} value modules.{module}.{dotted} must be a string")
        identifier = (module, dotted)
        if identifier in answers:
            raise Exception(f"--module-answers {source} defines modules.{module}.{dotted} more than once")
        answers[identifier] = value


def validate_module_answers(
    supplied: dict[tuple[str, str], str] | None,
    pending: tuple[ConfigQuestion, ...],
    *,
    source: Path | None = None,
) -> dict[tuple[str, str], str]:
    answers = supplied or {}
    if source is not None:
        source_label = f"--module-answers {source}"
    elif supplied is None:
        source_label = "no --module-answers file"
    else:
        source_label = "in-process module answers"
    for identifier, value in answers.items():
        if (
            not isinstance(identifier, tuple)
            or len(identifier) != 2
            or not all(isinstance(part, str) for part in identifier)
            or not isinstance(value, str)
        ):
            raise Exception(f"{source_label} must map (module, key) pairs to strings")
    expected = {(question.module, question.key) for question in pending}
    extra = sorted(set(answers) - expected)
    if extra:
        module, key = extra[0]
        raise Exception(f"{source_label} contains modules.{module}.{key}, which is not a pending question")
    missing = [question for question in pending if (question.module, question.key) not in answers]
    if missing and supplied is None:
        question = missing[0]
        raise Exception(
            f"pending question modules.{question.module}.{question.key} has no answer; "
            "pass --module-answers (run --list-config-questions first)"
        )
    if missing:
        question = missing[0]
        raise Exception(
            f"{source_label} is missing an answer for pending question "
            f"modules.{question.module}.{question.key}; run "
            "--list-config-questions first"
        )
    return answers


TABLE_HEADER = re.compile(r"\s*\[(?P<array>\[)?(?P<name>[^\[\]]+)\]\]?\s*(?:#.*)?\Z")


def text_with_answers(text: str | None, added: list[tuple[tuple[str, ...], str]], expected: dict) -> str:
    """The file's own text with the new answers added, so comments and layout survive.

    The result must parse to exactly the expected values. When it does not, or
    a table cannot be placed by text, the whole file is rendered instead.
    """
    if text is not None:
        try:
            for path, value in added:
                text = insert_answer(text, path, value)
            if tomllib.loads(text) == expected:
                return text
        except (ValueError, tomllib.TOMLDecodeError):
            pass
    return render_toml(expected)


def insert_answer(text: str, path: tuple[str, ...], value: str) -> str:
    table, leaf = path[:-1], path[-1]
    entry = f"{toml_key(leaf)} = {toml_string(value)}"
    lines = text.split("\n")
    headers = [(index, header_path(line)) for index, line in enumerate(lines) if TABLE_HEADER.match(line)]
    for position, (index, name) in enumerate(headers):
        if name != table:
            continue
        end = headers[position + 1][0] if position + 1 < len(headers) else len(lines)
        last = max(
            (row for row in range(index, end) if lines[row].strip() and not lines[row].lstrip().startswith("#")),
            default=index,
        )
        return "\n".join([*lines[: last + 1], entry, *lines[last + 1 :]])
    if lookup(tomllib.loads(text), table) is not _MISSING:
        raise ValueError(f"{'.'.join(table)} is not written as a table header")
    header = ".".join(toml_key(part) for part in table)
    separator = "" if not text or text.endswith("\n\n") else "\n" if text.endswith("\n") else "\n\n"
    return f"{text}{separator}[{header}]\n{entry}\n"


def header_path(line: str) -> tuple[str, ...] | None:
    match = TABLE_HEADER.match(line)
    if match is None or match.group("array"):
        return None
    try:
        data = tomllib.loads(f"{match.group('name')} = 1")
    except tomllib.TOMLDecodeError:
        return None
    parts: list[str] = []
    while isinstance(data, dict):
        ((key, data),) = data.items()
        parts.append(key)
    return tuple(parts)


def set_missing_value(
    data: dict,
    path: tuple[str, ...],
    value: str,
    source: Path,
) -> None:
    current = data
    for part in path[:-1]:
        child = current.get(part)
        if child is None:
            child = {}
            current[part] = child
        elif not isinstance(child, dict):
            dotted = ".".join(path)
            raise Exception(f"cannot add {dotted}: parent value in {source} is not a table")
        current = child
    leaf = path[-1]
    if leaf in current:
        raise Exception(f"refusing to overwrite existing {'.'.join(path)} in {source}")
    current[leaf] = value


def fill_team_config(text: str, project_root: Path) -> str:
    return text.replace("{directory_name}", project_root.name)


def output_folder(config_text: str) -> str:
    folder = tomllib.loads(config_text).get("core", {}).get("output_folder", "_bmad-output")
    prefix = "{project-root}/"
    if folder.startswith(prefix):
        folder = folder[len(prefix) :]
    return folder or "_bmad-output"


def materialize_bmad(
    project_root: Path,
    scripts_src: Path,
    config_text: str,
    module_trees: dict[str, PlainTree],
    *,
    user_config_text: str | None = None,
    custom_renames: tuple[tuple[str, str], ...] = (),
) -> None:
    bmad = project_root / "_bmad"
    project_root.mkdir(parents=True, exist_ok=True)
    # One attempt, not mkdtemp: on Windows, older Pythons' mkdtemp takes "access denied"
    # for a name collision and tries the next name, some two billion times. Setup
    # would hang in a folder it cannot write to instead of reporting the failure.
    staging = project_root / f"_bmad.setup-{os.urandom(8).hex()}"
    staging.mkdir()
    try:
        # Seed staging so custom/, extra *.user.toml, and leftovers
        # survive replace_dir.
        if bmad.exists():
            scripts = bmad / "scripts"

            def ignore_scripts_link(directory: str, _names: list[str]) -> set[str]:
                if scripts.is_symlink() and Path(directory) == bmad:
                    return {"scripts"}
                return set()

            shutil.copytree(
                bmad,
                staging,
                dirs_exist_ok=True,
                symlinks=True,
                ignore=ignore_scripts_link,
            )
        stage_bmad(
            staging,
            scripts_src=scripts_src,
            config_text=config_text,
            module_trees=module_trees,
            user_config_text=user_config_text,
            custom_renames=custom_renames,
        )
        replace_dir(staging, bmad)
    except Exception:
        shutil.rmtree(staging, ignore_errors=True)
        raise


def replace_dir(src: Path, dest: Path) -> None:
    if not dest.exists():
        src.rename(dest)
        return
    # Not mkdtemp: Windows refuses a rename onto an existing directory.
    backup = dest.with_name(f"_bmad.old-{datetime.datetime.now():%Y%m%d-%H%M%S}")
    dest.rename(backup)
    try:
        src.rename(dest)
    except Exception:
        backup.rename(dest)
        raise
    shutil.rmtree(backup)


def stage_bmad(
    staging: Path,
    *,
    scripts_src: Path,
    config_text: str,
    module_trees: dict[str, PlainTree],
    user_config_text: str | None,
    custom_renames: tuple[tuple[str, str], ...] = (),
) -> None:
    ensure_scripts(staging / "scripts", scripts_src)
    ensure_file(staging / "config.toml", config_text)
    # A module's scripts directory exists even when it declares no scripts,
    # so a second run reports current, not repaired.
    for code, tree in module_trees.items():
        ensure_plain_tree(staging / code / "scripts", tree)
    custom = staging / "custom"
    ensure_dir(custom)
    if custom.is_symlink() or not custom.is_dir():
        return
    gitignore = custom / ".gitignore"
    if not gitignore.exists() and not gitignore.is_symlink():
        write_text(gitignore, CUSTOM_GITIGNORE)
    if user_config_text is not None:
        write_text(custom / "config.user.toml", user_config_text)
    for old, new in custom_renames:
        (custom / old).rename(custom / new)


def ensure_scripts(dest: Path, src: Path) -> None:
    source = read_plain_tree(src)
    if tree_matches(dest, source):
        return
    if dest.is_symlink() or dest.is_file():
        dest.unlink()
    elif dest.exists():
        shutil.rmtree(dest)
    dest.mkdir(parents=True)
    for relative in source.directories:
        dest.joinpath(*relative.parts).mkdir(parents=True, exist_ok=True)
    for relative, _content in source.files:
        target = dest.joinpath(*relative.parts)
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(src.joinpath(*relative.parts), target)


def ensure_plain_tree(dest: Path, source: PlainTree) -> None:
    if tree_matches(dest, source):
        return
    if dest.is_symlink() or dest.is_file():
        dest.unlink()
    elif dest.exists():
        shutil.rmtree(dest)
    dest.mkdir(parents=True)
    for relative in source.directories:
        dest.joinpath(*relative.parts).mkdir(parents=True, exist_ok=True)
    for relative, content in source.files:
        target = dest.joinpath(*relative.parts)
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(content)


def write_text(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        content if content.endswith("\n") else content + "\n",
        encoding="utf-8",
    )


def ensure_file(path: Path, content: str) -> None:
    if path.is_symlink():
        path.unlink()
    elif path.is_file():
        existing = path.read_text(encoding="utf-8")
        filled = fill_toml(existing, content)
        if filled == existing:
            return
        content = filled
    elif path.exists():
        shutil.rmtree(path)
    write_text(path, content)


def ensure_dir(path: Path) -> None:
    if not path.is_symlink() and not path.exists():
        path.mkdir(parents=True)


def toml_string(value: str) -> str:
    replacements = {
        "\\": "\\\\",
        '"': '\\"',
        "\b": "\\b",
        "\t": "\\t",
        "\n": "\\n",
        "\f": "\\f",
        "\r": "\\r",
    }
    escaped = "".join(replacements.get(character, toml_control(character)) for character in value)
    return f'"{escaped}"'


def toml_control(value: str) -> str:
    codepoint = ord(value)
    if codepoint < 0x20 or codepoint == 0x7F:
        return f"\\u{codepoint:04X}"
    return value


def toml_key(key: str) -> str:
    if key and key.isascii() and key[0].isalpha() and all(c.isalnum() or c in "-_" for c in key):
        return key
    return toml_string(key)


def toml_value(value: object) -> str:
    if isinstance(value, str):
        return toml_string(value)
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, int):
        return str(value)
    if isinstance(value, float):
        return str(value)
    if isinstance(value, (datetime.datetime, datetime.date, datetime.time)):
        return value.isoformat()
    if value is None:
        return '""'
    if isinstance(value, list):
        return "[ " + ", ".join(toml_value(item) for item in value) + " ]"
    if isinstance(value, dict):
        rendered = ", ".join(f"{toml_key(str(key))} = {toml_value(item)}" for key, item in value.items())
        return "{ " + rendered + " }"
    return toml_string(str(value))


def fill_keep(template: object, existing: object) -> object:
    if isinstance(template, dict) and isinstance(existing, dict):
        result = dict(template)
        for key, value in existing.items():
            result[key] = fill_keep(result[key], value) if key in result else value
        return result
    return existing


def render_toml(data: dict) -> str:
    lines: list[str] = []

    def emit_scalars(table: dict) -> None:
        for key, value in table.items():
            if not isinstance(value, dict):
                lines.append(f"{toml_key(str(key))} = {toml_value(value)}")

    def emit_tables(table: dict, prefix: tuple[str, ...]) -> None:
        for key, value in table.items():
            if not isinstance(value, dict):
                continue
            header = (*prefix, str(key))
            scalars = any(not isinstance(item, dict) for item in value.values())
            nested = any(isinstance(item, dict) for item in value.values())
            if scalars or not nested:
                if lines:
                    lines.append("")
                lines.append(f"[{'.'.join(toml_key(part) for part in header)}]")
                emit_scalars(value)
            emit_tables(value, header)

    emit_scalars(data)
    emit_tables(data, ())
    return "\n".join(lines) + "\n"


def fill_toml(existing_text: str, template_text: str) -> str:
    try:
        existing = tomllib.loads(existing_text)
        template = tomllib.loads(template_text)
    except tomllib.TOMLDecodeError as error:
        raise Exception(f"cannot merge malformed TOML: {error}") from error
    if not isinstance(existing, dict) or not isinstance(template, dict):
        return template_text
    merged = fill_keep(template, existing)
    if merged == existing:
        return existing_text
    if merged == template:
        return template_text
    return render_toml(merged)


def cli() -> int:
    try:
        return main()
    except Exception as error:
        sys.stderr.write(f"error: {error}\n")
        return 1


if __name__ == "__main__":
    if sys.platform == "win32":
        # Piped output on Windows defaults to a legacy code page, not UTF-8.
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stderr.reconfigure(encoding="utf-8")
    raise SystemExit(cli())