From eda7f5444c0a5372f460b421caf4dc15672bd93e Mon Sep 17 00:00:00 2001
From: "alexandre.azouri" <alexandre.azouri@sogeti.com>
Date: Thu, 7 May 2026 16:59:36 +0200
Subject: [PATCH] feat: ajouter le module BMAD Orchestrator avec configuration
 et documentation

---
 .gitignore                                    |   4 +-
 orchestrator/bmad-orchestrator/SKILL.md       | 109 +++++
 .../bmad-orchestrator/assets/module-help.csv  |   2 +
 .../bmad-orchestrator/assets/module-setup.md  |  10 +
 .../bmad-orchestrator/assets/module.yaml      |   8 +
 orchestrator/bmad-orchestrator/customize.toml |  27 ++
 .../bmad-orchestrator/scripts/merge-config.py | 408 ++++++++++++++++++
 .../scripts/merge-help-csv.py                 | 218 ++++++++++
 8 files changed, 785 insertions(+), 1 deletion(-)
 create mode 100644 orchestrator/bmad-orchestrator/SKILL.md
 create mode 100644 orchestrator/bmad-orchestrator/assets/module-help.csv
 create mode 100644 orchestrator/bmad-orchestrator/assets/module-setup.md
 create mode 100644 orchestrator/bmad-orchestrator/assets/module.yaml
 create mode 100644 orchestrator/bmad-orchestrator/customize.toml
 create mode 100755 orchestrator/bmad-orchestrator/scripts/merge-config.py
 create mode 100755 orchestrator/bmad-orchestrator/scripts/merge-help-csv.py

diff --git a/.gitignore b/.gitignore
index c2fd8a370..0c1203b2c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -50,6 +50,7 @@ z*/
 
 _bmad
 _bmad-output
+skills/
 
 # Personal customization files (team files are committed, personal files are not)
 _bmad/custom/*.user.toml
@@ -84,4 +85,5 @@ build/
 
 
 # Local development files
-orchestrator/
\ No newline at end of file
+orchestrator/SKILL.md
+orchestrator/copilot-instructions.md
\ No newline at end of file
diff --git a/orchestrator/bmad-orchestrator/SKILL.md b/orchestrator/bmad-orchestrator/SKILL.md
new file mode 100644
index 000000000..01de79f55
--- /dev/null
+++ b/orchestrator/bmad-orchestrator/SKILL.md
@@ -0,0 +1,109 @@
+---
+name: bmad-orchestrator
+description: >
+  Agent BMAD autonome qui comprend l'intention utilisateur en langage naturel,
+  detecte les modules installes, route vers le bon skill/workflow et enchaine
+  les etapes sans que l'utilisateur ait a connaitre les commandes.
+  Use when the user asks to talk to the orchestrator.
+---
+
+# BMAD Orchestrator
+
+## Overview
+
+Agent d'orchestration intelligent pour BMAD. Il elimine la friction d'adoption en transformant toute intention exprimee en langage naturel en execution du bon skill, dans le bon ordre, avec un minimum de confirmations.
+
+## Identity
+
+Orchestrateur BMAD - routeur de workflows, facilitateur silencieux. Ne se met jamais en avant ; l'utilisateur percoit le resultat, pas la mecanique.
+
+## Communication Style
+
+Concis, oriente action. Repond en `{communication_language}`. Pas de bavardage - une phrase de contexte, puis l'action. Quand il propose des options, il les classe par pertinence avec une recommandation claire.
+
+## Principles
+
+- L'utilisateur ne devrait jamais avoir a memoriser un nom de skill ou un code.
+- Confirmations limitees a <= 10 % des interactions. En cas de doute, proposer 2-3 options classees plutot que bloquer.
+- La documentation BMAD locale est la source de verite. Le web est un fallback signale.
+- Respecter le choix utilisateur : forcer un skill, refuser le routage, ou ignorer une suggestion.
+- Ne jamais exfiltrer de donnees du repository hors de la session.
+
+---
+
+## On Activation
+
+1. **Charger la configuration** :
+   - Lire `{project-root}/_bmad/config.yaml` puis `{project-root}/_bmad/config.user.yaml`.
+   - Resoudre `{project_name}`, `{user_name}`, `{communication_language}` et `{output_folder}`.
+
+2. **Detecter les modules BMAD installes** :
+   - Lire `{project-root}/_bmad/_config/manifest.yaml` si present.
+   - Sinon, detecter les dossiers module sous `{project-root}/_bmad/`.
+
+3. **Charger le catalogue de skills** :
+   - Lire `{project-root}/_bmad/_config/skill-manifest.csv`.
+   - Lire `{project-root}/_bmad/_config/bmad-help.csv`.
+   - Filtrer les skills dont le module n'est pas installe.
+
+4. **Charger le contexte projet** :
+   - Chercher `**/project-context.md`. Si trouve, le charger comme reference.
+
+5. **Presenter le role** :
+   - Saluer `{user_name}` en `{communication_language}`.
+   - Expliquer en une phrase que l'orchestrateur route vers le bon workflow BMAD selon l'intention.
+
+---
+
+## Intent Routing Engine
+
+Quand l'utilisateur exprime une intention :
+
+### Etape 1 - Comprendre l'intention
+
+- Extraire l'objectif principal de la requete en langage naturel.
+- Mapper l'intention aux categories connues via `bmad-help.csv` et la description des skills installes.
+- Si l'intention est claire, router directement.
+- Si l'intention est ambigue, proposer au maximum 3 interpretations classees.
+
+### Etape 2 - Verifier les preconditions
+
+- Verifier si le skill cible est disponible dans le catalogue installe.
+- Verifier les prerequis de `bmad-help.csv` quand ils existent.
+- Si des artefacts requis manquent, proposer de lancer le prerequis d'abord.
+
+### Etape 3 - Executer
+
+- Invoquer le skill identifie par son nom exact.
+- Laisser le skill prendre le controle de la conversation.
+
+### Etape 4 - Suggestion proactive post-execution
+
+- Consulter `bmad-help.csv` pour identifier le skill suivant recommande.
+- Proposer la prochaine etape recommandee avec une justification en une phrase.
+
+---
+
+## User Controls
+
+### Forcer un skill
+
+Si l'utilisateur mentionne explicitement un nom de skill, un code ou un nom d'agent, router directement vers ce skill ou cet agent sans confirmation.
+
+### Bypass du routage
+
+Si l'utilisateur dit "pas de skill", "sans agent" ou "juste reponds" :
+- Repondre directement sans invoquer de skill.
+- Reprendre le routage automatique a la prochaine interaction.
+
+### Gouvernance d'equipe
+
+Si `{project-root}/_bmad/custom/config.user.toml` existe :
+- Lire les surcharges de configuration.
+- Appliquer ces regles pour la session.
+
+---
+
+## Capabilities
+
+L'orchestrateur ne possede pas de capabilities propres - il route vers les skills existants.
\ No newline at end of file
diff --git a/orchestrator/bmad-orchestrator/assets/module-help.csv b/orchestrator/bmad-orchestrator/assets/module-help.csv
new file mode 100644
index 000000000..6bfd70da2
--- /dev/null
+++ b/orchestrator/bmad-orchestrator/assets/module-help.csv
@@ -0,0 +1,2 @@
+module,skill,display-name,menu-code,description,action,args,phase,after,before,required,output-location,outputs
+BMAD Orchestrator,bmad-orchestrator,BMAD Orchestrator,OR,"Route natural-language intent to the right BMAD skill or named agent and suggests the next step.",orchestrate,"{freeform intent in natural language}",anytime,,,false,{project-root}/_bmad,_config/skill-manifest.csv
\ No newline at end of file
diff --git a/orchestrator/bmad-orchestrator/assets/module-setup.md b/orchestrator/bmad-orchestrator/assets/module-setup.md
new file mode 100644
index 000000000..b3a8f15fb
--- /dev/null
+++ b/orchestrator/bmad-orchestrator/assets/module-setup.md
@@ -0,0 +1,10 @@
+# Module Setup
+
+This standalone module does not require interactive setup prompts.
+
+On install:
+- Register module metadata from `assets/module.yaml`
+- Register help entries from `assets/module-help.csv`
+- Install `bmad-orchestrator` skill and its `customize.toml` agent metadata
+
+No additional user inputs are required.
\ No newline at end of file
diff --git a/orchestrator/bmad-orchestrator/assets/module.yaml b/orchestrator/bmad-orchestrator/assets/module.yaml
new file mode 100644
index 000000000..5b0118e24
--- /dev/null
+++ b/orchestrator/bmad-orchestrator/assets/module.yaml
@@ -0,0 +1,8 @@
+code: bmo
+name: "BMAD Orchestrator"
+description: "Standalone BMAD orchestrator agent module"
+module_version: 1.0.0
+default_selected: false
+module_greeting: >
+  BMAD Orchestrator est installe.
+  Vous pouvez maintenant router vos intentions BMAD en langage naturel.
diff --git a/orchestrator/bmad-orchestrator/customize.toml b/orchestrator/bmad-orchestrator/customize.toml
new file mode 100644
index 000000000..8adb68fa9
--- /dev/null
+++ b/orchestrator/bmad-orchestrator/customize.toml
@@ -0,0 +1,27 @@
+[agent]
+name = "Orchestrator"
+title = "BMAD Orchestrator"
+icon = "🧭"
+
+activation_steps_prepend = []
+activation_steps_append = []
+
+persistent_facts = [
+  "file:{project-root}/**/project-context.md",
+]
+
+role = "Route les intentions utilisateur vers le bon workflow ou agent BMAD avec le moins de friction possible."
+identity = "Facilitateur silencieux centre sur le routage, les preconditions et l'enchainement des etapes BMAD."
+communication_style = "Concis, oriente action, sans bavardage. Une phrase de contexte puis l'action."
+
+principles = [
+  "L'utilisateur ne devrait jamais avoir a memoriser un nom de skill ou un code.",
+  "Preferer le routage direct quand l'intention est claire.",
+  "Limiter les confirmations au strict minimum.",
+  "La documentation BMAD locale est la source de verite.",
+]
+
+[[agent.menu]]
+code = "OR"
+description = "Router une intention utilisateur vers le bon skill ou agent BMAD"
+prompt = "Analyse l'intention utilisateur, identifie le workflow BMAD ou l'agent approprie, verifie les preconditions si necessaire, puis poursuis avec ce skill ou cette reponse."
\ No newline at end of file
diff --git a/orchestrator/bmad-orchestrator/scripts/merge-config.py b/orchestrator/bmad-orchestrator/scripts/merge-config.py
new file mode 100755
index 000000000..6ee0ac7c2
--- /dev/null
+++ b/orchestrator/bmad-orchestrator/scripts/merge-config.py
@@ -0,0 +1,408 @@
+#!/usr/bin/env python3
+# /// script
+# requires-python = ">=3.9"
+# dependencies = ["pyyaml"]
+# ///
+"""Merge module configuration into shared _bmad/config.yaml and config.user.yaml.
+
+Reads a module.yaml definition and a JSON answers file, then writes or updates
+the shared config.yaml (core values at root + module section) and config.user.yaml
+(user_name, communication_language, plus any module variable with user_setting: true).
+Uses an anti-zombie pattern for the module section in config.yaml.
+
+Legacy migration: when --legacy-dir is provided, reads old per-module config files
+from {legacy-dir}/{module-code}/config.yaml and {legacy-dir}/core/config.yaml.
+Matching values serve as fallback defaults (answers override them). After a
+successful merge, the legacy config.yaml files are deleted. Only the current
+module and core directories are touched — other module directories are left alone.
+
+Exit codes: 0=success, 1=validation error, 2=runtime error
+"""
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+try:
+    import yaml
+except ImportError:
+    print("Error: pyyaml is required (PEP 723 dependency)", file=sys.stderr)
+    sys.exit(2)
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description="Merge module config into shared _bmad/config.yaml with anti-zombie pattern."
+    )
+    parser.add_argument(
+        "--config-path",
+        required=True,
+        help="Path to the target _bmad/config.yaml file",
+    )
+    parser.add_argument(
+        "--module-yaml",
+        required=True,
+        help="Path to the module.yaml definition file",
+    )
+    parser.add_argument(
+        "--answers",
+        required=True,
+        help="Path to JSON file with collected answers",
+    )
+    parser.add_argument(
+        "--user-config-path",
+        required=True,
+        help="Path to the target _bmad/config.user.yaml file",
+    )
+    parser.add_argument(
+        "--legacy-dir",
+        help="Path to _bmad/ directory to check for legacy per-module config files. "
+        "Matching values are used as fallback defaults, then legacy files are deleted.",
+    )
+    parser.add_argument(
+        "--verbose",
+        action="store_true",
+        help="Print detailed progress to stderr",
+    )
+    return parser.parse_args()
+
+
+def load_yaml_file(path: str) -> dict:
+    """Load a YAML file, returning empty dict if file doesn't exist."""
+    file_path = Path(path)
+    if not file_path.exists():
+        return {}
+    with open(file_path, "r", encoding="utf-8") as f:
+        content = yaml.safe_load(f)
+    return content if content else {}
+
+
+def load_json_file(path: str) -> dict:
+    """Load a JSON file."""
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+# Keys that live at config root (shared across all modules)
+_CORE_KEYS = frozenset(
+    {"user_name", "communication_language", "document_output_language", "output_folder"}
+)
+
+
+def load_legacy_values(
+    legacy_dir: str, module_code: str, module_yaml: dict, verbose: bool = False
+) -> tuple[dict, dict, list]:
+    """Read legacy per-module config files and return core/module value dicts.
+
+    Reads {legacy_dir}/core/config.yaml and {legacy_dir}/{module_code}/config.yaml.
+    Only returns values whose keys match the current schema (core keys or module.yaml
+    variable definitions). Other modules' directories are not touched.
+
+    Returns:
+        (legacy_core, legacy_module, files_found) where files_found lists paths read.
+    """
+    legacy_core: dict = {}
+    legacy_module: dict = {}
+    files_found: list = []
+
+    # Read core legacy config
+    core_path = Path(legacy_dir) / "core" / "config.yaml"
+    if core_path.exists():
+        core_data = load_yaml_file(str(core_path))
+        files_found.append(str(core_path))
+        for k, v in core_data.items():
+            if k in _CORE_KEYS:
+                legacy_core[k] = v
+        if verbose:
+            print(f"Legacy core config: {list(legacy_core.keys())}", file=sys.stderr)
+
+    # Read module legacy config
+    mod_path = Path(legacy_dir) / module_code / "config.yaml"
+    if mod_path.exists():
+        mod_data = load_yaml_file(str(mod_path))
+        files_found.append(str(mod_path))
+        for k, v in mod_data.items():
+            if k in _CORE_KEYS:
+                # Core keys duplicated in module config — only use if not already set
+                if k not in legacy_core:
+                    legacy_core[k] = v
+            elif k in module_yaml and isinstance(module_yaml[k], dict):
+                # Module-specific key that matches a current variable definition
+                legacy_module[k] = v
+        if verbose:
+            print(
+                f"Legacy module config: {list(legacy_module.keys())}", file=sys.stderr
+            )
+
+    return legacy_core, legacy_module, files_found
+
+
+def apply_legacy_defaults(answers: dict, legacy_core: dict, legacy_module: dict) -> dict:
+    """Apply legacy values as fallback defaults under the answers.
+
+    Legacy values fill in any key not already present in answers.
+    Explicit answers always win.
+    """
+    merged = dict(answers)
+
+    if legacy_core:
+        core = merged.get("core", {})
+        filled_core = dict(legacy_core)  # legacy as base
+        filled_core.update(core)  # answers override
+        merged["core"] = filled_core
+
+    if legacy_module:
+        mod = merged.get("module", {})
+        filled_mod = dict(legacy_module)  # legacy as base
+        filled_mod.update(mod)  # answers override
+        merged["module"] = filled_mod
+
+    return merged
+
+
+def cleanup_legacy_configs(
+    legacy_dir: str, module_code: str, verbose: bool = False
+) -> list:
+    """Delete legacy config.yaml files for this module and core only.
+
+    Returns list of deleted file paths.
+    """
+    deleted = []
+    for subdir in (module_code, "core"):
+        legacy_path = Path(legacy_dir) / subdir / "config.yaml"
+        if legacy_path.exists():
+            if verbose:
+                print(f"Deleting legacy config: {legacy_path}", file=sys.stderr)
+            legacy_path.unlink()
+            deleted.append(str(legacy_path))
+    return deleted
+
+
+def extract_module_metadata(module_yaml: dict) -> dict:
+    """Extract non-variable metadata fields from module.yaml."""
+    meta = {}
+    for k in ("name", "description"):
+        if k in module_yaml:
+            meta[k] = module_yaml[k]
+    meta["version"] = module_yaml.get("module_version")  # null if absent
+    if "default_selected" in module_yaml:
+        meta["default_selected"] = module_yaml["default_selected"]
+    return meta
+
+
+def apply_result_templates(
+    module_yaml: dict, module_answers: dict, verbose: bool = False
+) -> dict:
+    """Apply result templates from module.yaml to transform raw answer values.
+
+    For each answer, if the corresponding variable definition in module.yaml has
+    a 'result' field, replaces {value} in that template with the answer. Skips
+    the template if the answer already contains '{project-root}' to prevent
+    double-prefixing.
+    """
+    transformed = {}
+    for key, value in module_answers.items():
+        var_def = module_yaml.get(key)
+        if (
+            isinstance(var_def, dict)
+            and "result" in var_def
+            and "{project-root}" not in str(value)
+        ):
+            template = var_def["result"]
+            transformed[key] = template.replace("{value}", str(value))
+            if verbose:
+                print(
+                    f"Applied result template for '{key}': {value} → {transformed[key]}",
+                    file=sys.stderr,
+                )
+        else:
+            transformed[key] = value
+    return transformed
+
+
+def merge_config(
+    existing_config: dict,
+    module_yaml: dict,
+    answers: dict,
+    verbose: bool = False,
+) -> dict:
+    """Merge answers into config, applying anti-zombie pattern.
+
+    Args:
+        existing_config: Current config.yaml contents (may be empty)
+        module_yaml: The module definition
+        answers: JSON with 'core' and/or 'module' keys
+        verbose: Print progress to stderr
+
+    Returns:
+        Updated config dict ready to write
+    """
+    config = dict(existing_config)
+    module_code = module_yaml.get("code")
+
+    if not module_code:
+        print("Error: module.yaml must have a 'code' field", file=sys.stderr)
+        sys.exit(1)
+
+    # Migrate legacy core: section to root
+    if "core" in config and isinstance(config["core"], dict):
+        if verbose:
+            print("Migrating legacy 'core' section to root", file=sys.stderr)
+        config.update(config.pop("core"))
+
+    # Strip user-only keys from config — they belong exclusively in config.user.yaml
+    for key in _CORE_USER_KEYS:
+        if key in config:
+            if verbose:
+                print(f"Removing user-only key '{key}' from config (belongs in config.user.yaml)", file=sys.stderr)
+            del config[key]
+
+    # Write core values at root (global properties, not nested under "core")
+    # Exclude user-only keys — those belong exclusively in config.user.yaml
+    core_answers = answers.get("core")
+    if core_answers:
+        shared_core = {k: v for k, v in core_answers.items() if k not in _CORE_USER_KEYS}
+        if shared_core:
+            if verbose:
+                print(f"Writing core config at root: {list(shared_core.keys())}", file=sys.stderr)
+            config.update(shared_core)
+
+    # Anti-zombie: remove existing module section
+    if module_code in config:
+        if verbose:
+            print(
+                f"Removing existing '{module_code}' section (anti-zombie)",
+                file=sys.stderr,
+            )
+        del config[module_code]
+
+    # Build module section: metadata + variable values
+    module_section = extract_module_metadata(module_yaml)
+    module_answers = apply_result_templates(
+        module_yaml, answers.get("module", {}), verbose
+    )
+    module_section.update(module_answers)
+
+    if verbose:
+        print(
+            f"Writing '{module_code}' section with keys: {list(module_section.keys())}",
+            file=sys.stderr,
+        )
+
+    config[module_code] = module_section
+
+    return config
+
+
+# Core keys that are always written to config.user.yaml
+_CORE_USER_KEYS = ("user_name", "communication_language")
+
+
+def extract_user_settings(module_yaml: dict, answers: dict) -> dict:
+    """Collect settings that belong in config.user.yaml.
+
+    Includes user_name and communication_language from core answers, plus any
+    module variable whose definition contains user_setting: true.
+    """
+    user_settings = {}
+
+    core_answers = answers.get("core", {})
+    for key in _CORE_USER_KEYS:
+        if key in core_answers:
+            user_settings[key] = core_answers[key]
+
+    module_answers = answers.get("module", {})
+    for var_name, var_def in module_yaml.items():
+        if isinstance(var_def, dict) and var_def.get("user_setting") is True:
+            if var_name in module_answers:
+                user_settings[var_name] = module_answers[var_name]
+
+    return user_settings
+
+
+def write_config(config: dict, config_path: str, verbose: bool = False) -> None:
+    """Write config dict to YAML file, creating parent dirs as needed."""
+    path = Path(config_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+
+    if verbose:
+        print(f"Writing config to {path}", file=sys.stderr)
+
+    with open(path, "w", encoding="utf-8") as f:
+        yaml.dump(
+            config,
+            f,
+            default_flow_style=False,
+            allow_unicode=True,
+            sort_keys=False,
+        )
+
+
+def main():
+    args = parse_args()
+
+    # Load inputs
+    module_yaml = load_yaml_file(args.module_yaml)
+    if not module_yaml:
+        print(f"Error: Could not load module.yaml from {args.module_yaml}", file=sys.stderr)
+        sys.exit(1)
+
+    answers = load_json_file(args.answers)
+    existing_config = load_yaml_file(args.config_path)
+
+    if args.verbose:
+        exists = Path(args.config_path).exists()
+        print(f"Config file exists: {exists}", file=sys.stderr)
+        if exists:
+            print(f"Existing sections: {list(existing_config.keys())}", file=sys.stderr)
+
+    # Legacy migration: read old per-module configs as fallback defaults
+    legacy_files_found = []
+    if args.legacy_dir:
+        module_code = module_yaml.get("code", "")
+        legacy_core, legacy_module, legacy_files_found = load_legacy_values(
+            args.legacy_dir, module_code, module_yaml, args.verbose
+        )
+        if legacy_core or legacy_module:
+            answers = apply_legacy_defaults(answers, legacy_core, legacy_module)
+            if args.verbose:
+                print("Applied legacy values as fallback defaults", file=sys.stderr)
+
+    # Merge and write config.yaml
+    updated_config = merge_config(existing_config, module_yaml, answers, args.verbose)
+    write_config(updated_config, args.config_path, args.verbose)
+
+    # Merge and write config.user.yaml
+    user_settings = extract_user_settings(module_yaml, answers)
+    existing_user_config = load_yaml_file(args.user_config_path)
+    updated_user_config = dict(existing_user_config)
+    updated_user_config.update(user_settings)
+    if user_settings:
+        write_config(updated_user_config, args.user_config_path, args.verbose)
+
+    # Legacy cleanup: delete old per-module config files
+    legacy_deleted = []
+    if args.legacy_dir:
+        legacy_deleted = cleanup_legacy_configs(
+            args.legacy_dir, module_yaml["code"], args.verbose
+        )
+
+    # Output result summary as JSON
+    module_code = module_yaml["code"]
+    result = {
+        "status": "success",
+        "config_path": str(Path(args.config_path).resolve()),
+        "user_config_path": str(Path(args.user_config_path).resolve()),
+        "module_code": module_code,
+        "core_updated": bool(answers.get("core")),
+        "module_keys": list(updated_config.get(module_code, {}).keys()),
+        "user_keys": list(user_settings.keys()),
+        "legacy_configs_found": legacy_files_found,
+        "legacy_configs_deleted": legacy_deleted,
+    }
+    print(json.dumps(result, indent=2))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/orchestrator/bmad-orchestrator/scripts/merge-help-csv.py b/orchestrator/bmad-orchestrator/scripts/merge-help-csv.py
new file mode 100755
index 000000000..6ba1afe04
--- /dev/null
+++ b/orchestrator/bmad-orchestrator/scripts/merge-help-csv.py
@@ -0,0 +1,218 @@
+#!/usr/bin/env python3
+# /// script
+# requires-python = ">=3.9"
+# dependencies = []
+# ///
+"""Merge module help entries into shared _bmad/module-help.csv.
+
+Reads a source CSV with module help entries and merges them into a target CSV.
+Uses an anti-zombie pattern: all existing rows matching the source module code
+are removed before appending fresh rows.
+
+Legacy cleanup: when --legacy-dir and --module-code are provided, deletes old
+per-module module-help.csv files from {legacy-dir}/{module-code}/ and
+{legacy-dir}/core/. Only the current module and core are touched.
+
+Exit codes: 0=success, 1=validation error, 2=runtime error
+"""
+
+import argparse
+import csv
+import json
+import sys
+from io import StringIO
+from pathlib import Path
+
+# CSV header for module-help.csv
+HEADER = [
+    "module",
+    "skill",
+    "display-name",
+    "menu-code",
+    "description",
+    "action",
+    "args",
+    "phase",
+    "after",
+    "before",
+    "required",
+    "output-location",
+    "outputs",
+]
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description="Merge module help entries into shared _bmad/module-help.csv with anti-zombie pattern."
+    )
+    parser.add_argument(
+        "--target",
+        required=True,
+        help="Path to the target _bmad/module-help.csv file",
+    )
+    parser.add_argument(
+        "--source",
+        required=True,
+        help="Path to the source module-help.csv with entries to merge",
+    )
+    parser.add_argument(
+        "--legacy-dir",
+        help="Path to _bmad/ directory to check for legacy per-module CSV files.",
+    )
+    parser.add_argument(
+        "--module-code",
+        help="Module code (required with --legacy-dir for scoping cleanup).",
+    )
+    parser.add_argument(
+        "--verbose",
+        action="store_true",
+        help="Print detailed progress to stderr",
+    )
+    return parser.parse_args()
+
+
+def read_csv_rows(path: str) -> tuple[list[str], list[list[str]]]:
+    """Read CSV file returning (header, data_rows).
+
+    Returns empty header and rows if file doesn't exist.
+    """
+    file_path = Path(path)
+    if not file_path.exists():
+        return [], []
+
+    with open(file_path, "r", encoding="utf-8", newline="") as f:
+        content = f.read()
+
+    reader = csv.reader(StringIO(content))
+    rows = list(reader)
+
+    if not rows:
+        return [], []
+
+    return rows[0], rows[1:]
+
+
+def extract_module_codes(rows: list[list[str]]) -> set[str]:
+    """Extract unique module codes from data rows."""
+    codes = set()
+    for row in rows:
+        if row and row[0].strip():
+            codes.add(row[0].strip())
+    return codes
+
+
+def filter_rows(rows: list[list[str]], module_code: str) -> list[list[str]]:
+    """Remove all rows matching the given module code."""
+    return [row for row in rows if not row or row[0].strip() != module_code]
+
+
+def write_csv(path: str, header: list[str], rows: list[list[str]], verbose: bool = False) -> None:
+    """Write header + rows to CSV file, creating parent dirs as needed."""
+    file_path = Path(path)
+    file_path.parent.mkdir(parents=True, exist_ok=True)
+
+    if verbose:
+        print(f"Writing {len(rows)} data rows to {path}", file=sys.stderr)
+
+    with open(file_path, "w", encoding="utf-8", newline="") as f:
+        writer = csv.writer(f)
+        writer.writerow(header)
+        for row in rows:
+            writer.writerow(row)
+
+
+def cleanup_legacy_csvs(
+    legacy_dir: str, module_code: str, verbose: bool = False
+) -> list:
+    """Delete legacy per-module module-help.csv files for this module and core only.
+
+    Returns list of deleted file paths.
+    """
+    deleted = []
+    for subdir in (module_code, "core"):
+        legacy_path = Path(legacy_dir) / subdir / "module-help.csv"
+        if legacy_path.exists():
+            if verbose:
+                print(f"Deleting legacy CSV: {legacy_path}", file=sys.stderr)
+            legacy_path.unlink()
+            deleted.append(str(legacy_path))
+    return deleted
+
+
+def main():
+    args = parse_args()
+
+    # Read source entries
+    source_header, source_rows = read_csv_rows(args.source)
+    if not source_rows:
+        print(f"Error: No data rows found in source {args.source}", file=sys.stderr)
+        sys.exit(1)
+
+    # Determine module codes being merged
+    source_codes = extract_module_codes(source_rows)
+    if not source_codes:
+        print("Error: Could not determine module code from source rows", file=sys.stderr)
+        sys.exit(1)
+
+    if args.verbose:
+        print(f"Source module codes: {source_codes}", file=sys.stderr)
+        print(f"Source rows: {len(source_rows)}", file=sys.stderr)
+
+    # Read existing target (may not exist)
+    target_header, target_rows = read_csv_rows(args.target)
+    target_existed = Path(args.target).exists()
+
+    if args.verbose:
+        print(f"Target exists: {target_existed}", file=sys.stderr)
+        if target_existed:
+            print(f"Existing target rows: {len(target_rows)}", file=sys.stderr)
+
+    # Use source header if target doesn't exist or has no header
+    header = target_header if target_header else (source_header if source_header else HEADER)
+
+    # Anti-zombie: remove all rows for each source module code
+    filtered_rows = target_rows
+    removed_count = 0
+    for code in source_codes:
+        before_count = len(filtered_rows)
+        filtered_rows = filter_rows(filtered_rows, code)
+        removed_count += before_count - len(filtered_rows)
+
+    if args.verbose and removed_count > 0:
+        print(f"Removed {removed_count} existing rows (anti-zombie)", file=sys.stderr)
+
+    # Append source rows
+    merged_rows = filtered_rows + source_rows
+
+    # Write result
+    write_csv(args.target, header, merged_rows, args.verbose)
+
+    # Legacy cleanup: delete old per-module CSV files
+    legacy_deleted = []
+    if args.legacy_dir:
+        if not args.module_code:
+            print(
+                "Error: --module-code is required when --legacy-dir is provided",
+                file=sys.stderr,
+            )
+            sys.exit(1)
+        legacy_deleted = cleanup_legacy_csvs(
+            args.legacy_dir, args.module_code, args.verbose
+        )
+
+    # Output result summary as JSON
+    result = {
+        "status": "success",
+        "target_path": str(Path(args.target).resolve()),
+        "target_existed": target_existed,
+        "module_codes": sorted(source_codes),
+        "rows_removed": removed_count,
+        "rows_added": len(source_rows),
+        "total_rows": len(merged_rows),
+        "legacy_csvs_deleted": legacy_deleted,
+    }
+    print(json.dumps(result, indent=2))
+
+
+if __name__ == "__main__":
+    main()