bundle_skill.py

#!/usr/bin/env python3
"""Assemble the Claude skill bundle(s) for the `mage` plugin from the catalogue sources.

A Claude plugin is copied to a local cache on install and cannot reach files outside its own directory,
so each skill must vendor everything it references. This script is the single writer of the generated
half of the plugin, and it is **registry-driven**: each `SkillSpec` in `SPECS` declares what one skill
needs, and the same machinery bundles every spec — so adding a skill is "add a `SkillSpec`," not a fork.

For a skill that mirrors the catalogue (e.g. `self-governance`), it copies the requested roles' entries
into `alignment/mechanisms/`, emits a scoped `INDEX.md`, vendors any referenced download templates, and lifts Part A
of a starter governance doc into `principles.md`. A skill with no mirror (`include_roles=()`) gets just
its `principles.md` — its own reference material ships in the authored `SKILL.md` half.

It is the skill analogue of `catalog.py build`: the bundles are GENERATED, never hand-edited, so they
can't drift from the catalogue. Runs on every commit via `hooks/pre-commit`. Stdlib-only; imports
`catalog` for the shared entry model.

Two modes, one registry (`SPECS`):

  * **build** (no args) — the generation half above: regenerate every GENERATED skill's `principles.md`,
    `alignment/mechanisms/`, and `templates/` from the catalogue sources. This is what the pre-commit hook runs.
  * **install / refresh** (`--install SKILL DEST` / `--refresh SKILL DEST`) — the ADOPTER half: copy a
    managed skill out of the plugin (the upstream SSOT) into an adopter's skills dir, under a **disjoint
    partition**. Every installed file is owned by exactly one side: *upstream-owned* (anything NOT named
    `*.local.md` and NOT under a `local/` directory) is overwritten on every refresh; *adopter-owned*
    (`*.local.md` files + everything under `local/`) is never read, written, or deleted. The two sets are
    disjoint by construction, so **refresh is a pure overwrite-preserve — no 3-way merge exists to get
    wrong**. An `.upstream-manifest` at the dest records the upstream-owned set so refresh can retire a
    file that vanished upstream without ever touching an adopter file.

Exit codes (per the repo's subprocess convention): 0 = success, 1 = failure.
"""
from __future__ import annotations

import argparse
import glob
import os
import re
import shutil
import sys
from typing import NamedTuple

import catalog  # shared Entry model + ROOT; import-safe (main() is __main__-guarded)

ROOT = catalog.ROOT
PLUGIN_SKILLS = os.path.join(ROOT, "plugin", "mage", "skills")  # the umbrella plugin's skills dir
PUBLIC = "https://davisjam.github.io/model-based-agentic-software-engineering/"  # base for links that leave the bundle

GEN_NOTE = "<!-- GENERATED by bundle_skill.py from the catalogue sources — do not hand-edit; edits are overwritten on the next build. -->\n"


class SkillSpec(NamedTuple):
    """One skill's bundle recipe. Adding a skill = adding a `SkillSpec` to `SPECS` below.

    A spec drives TWO registries. The **generation** half (`build`) runs only for a spec that has a
    `starter_doc` (see `is_generated`) — it lifts Part A into `principles.md` and, with `include_roles`,
    mirrors the catalogue. A spec with no `starter_doc` (a purely HAND-AUTHORED skill, e.g. self-communicate)
    is generation-inert but still a first-class member of the **install/refresh** managed set: it is
    installable into an adopter and its `*.local.md` / `local/` overlay is honored. Registering such a skill
    here is what ends its hand-drift — one SSOT (its plugin dir), one install path, no un-managed mirror.
    """
    name: str                 # the skill dir name under the plugin's skills/
    starter_doc: str = ""     # abs path to the Part-A source doc (Part A → principles.md); "" = not generated
    include_roles: tuple = ()  # catalogue roles mirrored into alignment/mechanisms/ ; () = no mirror
    principles_intro: str = ""  # preamble prepended to the lifted Part A in principles.md
    vendor_downloads: tuple = ()  # downloads/ templates shipped in <skill>/templates/ — explicit, so a
    #                               skill with no bundled entries (no reference-based vendoring) still gets them

    @property
    def skill_dir(self) -> str:
        return os.path.join(PLUGIN_SKILLS, self.name)

    @property
    def mechanisms_dir(self) -> str:
        # the GENERATED census now lives BENEATH the hand-authored alignment/ facet
        return os.path.join(self.skill_dir, "alignment", "mechanisms")

    @property
    def is_generated(self) -> bool:
        """A spec is generated (emits principles.md / a mirror) iff it names a Part-A starter doc. A
        hand-authored, install-only skill has no starter doc and is skipped by the `build` mode."""
        return bool(self.starter_doc)


# ── The skill registry ─────────────────────────────────────────────────────────────────────────────
SELF_GOVERNANCE = SkillSpec(
    name="self-governance",
    starter_doc=os.path.join(ROOT, "downloads", "CLAUDE-starter.md"),
    include_roles=("agent", "models-bridge"),  # product/ excluded: it governs the artifact, not the self
    principles_intro=(
        GEN_NOTE
        + "<!-- Source: Part A of the project's starter governance doc (a portable mirror of a "
        "mature CLAUDE.md). The copyright / attribution header below is retained verbatim. -->\n\n"
        "# Operating principles — the Davis AI-First Engineering Method\n\n"
        "*These portable principles are the operating stance of the `self-governance` skill. "
        "The catalogue in `alignment/mechanisms/` is the set of concrete controls; these principles are the "
        "reasoning that decides when to reach for one.*\n\n---\n\n"
    ),
    # The model-based-software-engineering starter kit: fill-in scaffolds for authoring the 4+1
    # model views (the catalogue documents them as mechanisms; these let a fresh project BUILD them).
    # Plus the shared method-language glossary — the lexicon the govern/operate/communicate trio speaks
    # in, so a term means the same thing across all three skills.
    vendor_downloads=(
        "method-language-starter.md",
        "system-models-starter-kit.md",
        "state-machine-model-starter.py",
        "component-zone-model-starter.py",
        "service-flow-model-starter.yaml",
        "service-flow-loader-starter.py",
        "deployment-topology-starter.py",
    ),
)

SELF_OPERATIONS = SkillSpec(
    name="self-operations",
    starter_doc=os.path.join(ROOT, "downloads", "self-operations-starter.md"),
    include_roles=(),  # no catalogue mirror — its reference is the authored SKILL.md map + examples/
    principles_intro=(
        GEN_NOTE
        + "<!-- Source: Part A of the self-operations starter (the portable operating mindset). -->\n\n"
        "# Operating principles — the self-operations mindset\n\n"
        "*The operating stance of the `self-operations` skill: how to run an agent-fleet substrate as a "
        "DevOps engineer. The skill's `SKILL.md` carries the lifecycle map and the bootstrap; these "
        "principles are the mindset you operate with. Partner to the `self-governance` skill.*\n\n---\n\n"
    ),
    # The operate skill authors design docs + Epics (infra changes, the feature-dev flow) — ship the
    # templates locally so it never has to reach outside its own dir (self-governance already vendors
    # these via its alignment/mechanisms/downloads/ mirror).
    vendor_downloads=(
        "EPIC-TEMPLATE-starter.md",
        "design-doc-template-starter.md",
        "emit-handoff-starter.py",  # machine-generate the reconstructable handoff sections (L2 banking).
        # The Part-B substrate: two typed SSOTs + a combined generator/ref-lint, so the bootstrap's
        # "emit Part B" step is scaffolded machinery, not prose (the SSOT→generator→ref-lint triangle).
        "pointers-starter.yaml",
        "runbooks-starter.yaml",
        "gen-and-lint-partb-starter.py",
    ),
)

SELF_COMMUNICATE = SkillSpec(
    name="self-communicate",
    # Hand-authored — no Part-A starter doc, no catalogue mirror, so `is_generated` is False and the
    # `build` mode skips it. It is registered here to bring it under the install/refresh MANAGED set: its
    # plugin dir is the single SSOT, the install path propagates it to an adopter, and its
    # writing/lexicon.local.md overlay is honored. Before this, self-communicate had two hand-synced copies
    # (a parent-repo mirror + this plugin copy) that DIFFERED and drifted — the exact mirror-drift class the
    # Epic kills. One authored copy + one managed install path retires that class (Phase-1b review §3.7).
)

SPECS = [SELF_GOVERNANCE, SELF_OPERATIONS, SELF_COMMUNICATE]


def _write(path: str, text: str) -> None:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(text)


def _write_copy(src: str, dst: str) -> None:
    os.makedirs(os.path.dirname(dst), exist_ok=True)
    shutil.copyfile(src, dst)


def _mirror_entries(spec: SkillSpec) -> int:
    """Copy each included role's entries + role README verbatim into alignment/mechanisms/."""
    for role in spec.include_roles:
        dst = os.path.join(spec.mechanisms_dir, role)
        if os.path.isdir(dst):
            shutil.rmtree(dst)
    n = 0
    for e in catalog.all_entries():
        role = e.path.split(os.sep)[0]
        if role not in spec.include_roles:
            continue
        dst = os.path.join(spec.mechanisms_dir, e.path)
        os.makedirs(os.path.dirname(dst), exist_ok=True)
        shutil.copyfile(os.path.join(ROOT, e.path), dst)
        n += 1
    for role in spec.include_roles:  # role READMEs aren't entries but ground the family structure
        src = os.path.join(ROOT, role, "README.md")
        if os.path.isfile(src):
            shutil.copyfile(src, os.path.join(spec.mechanisms_dir, role, "README.md"))
    shutil.copyfile(os.path.join(ROOT, "ABSTRACTIONS.md"), os.path.join(spec.mechanisms_dir, "ABSTRACTIONS.md"))
    return n


def _filtered_index(spec: SkillSpec) -> None:
    """Emit alignment/mechanisms/INDEX.md = the census truncated to the included roles."""
    full = open(os.path.join(ROOT, "INDEX.md"), encoding="utf-8").read()
    cut = full.find("# Product target")  # everything above product is agent + models-bridge
    body = full[:cut].rstrip() if cut != -1 else full.rstrip()
    # the source line says "All 53 entries…" (full catalogue); this bundle ships only agent+bridge, so
    # rewrite the count to the bundled total to avoid a 53-vs-33 confusion for a fresh reader.
    n_bundled = sum(1 for e in catalog.all_entries() if e.path.split(os.sep)[0] in spec.include_roles)
    body = re.sub(r"All \d+ entries are fully\s+written[^.]*\.",
                  f"All {n_bundled} bundled entries are fully written "
                  "(the product role's entries live in the full catalogue).", body)
    header = (
        GEN_NOTE
        + "<!-- Scope: the agent + models-bridge roles only. The product role is "
        "excluded from this skill; see the full catalogue for it. -->\n\n"
    )
    _write(os.path.join(spec.mechanisms_dir, "INDEX.md"), header + body + "\n")


def _principles(spec: SkillSpec) -> None:
    """Lift Part A (the portable method) out of the skill's starter doc into principles.md."""
    txt = open(spec.starter_doc, encoding="utf-8").read()
    a = txt.find("# Part A")
    b = txt.find("# Part B")
    if a == -1 or b == -1:
        raise SystemExit(f"bundle_skill: could not locate Part A / Part B markers in {spec.starter_doc}")
    part_a = txt[a:b].rstrip().rstrip("-").rstrip()  # drop the trailing '---' separator
    _write(os.path.join(spec.skill_dir, "principles.md"), spec.principles_intro + part_a + "\n")


def _reference_readme(spec: SkillSpec, n: int) -> None:
    _write(
        os.path.join(spec.mechanisms_dir, "README.md"),
        GEN_NOTE
        + "\n# Reference catalogue (bundled)\n\n"
        f"A vendored copy of the **agent** and **models-bridge** roles of the "
        f"agent-governance-mechanisms catalogue ({n} mechanism entries), assembled by "
        "`bundle_skill.py` so the skill is self-contained when installed. Start at "
        "[`INDEX.md`](INDEX.md); read individual entries on demand. Canonical source + the "
        "product role: https://davisjam.github.io/model-based-agentic-software-engineering/\n",
    )


def _vendor_referenced_downloads(spec: SkillSpec) -> int:
    """Copy any downloads/ template a BUNDLED entry references into alignment/mechanisms/downloads/, so the
    'Adopt it ->' handoff links resolve when the skill is installed (the plugin cache can't reach
    outside its own dir). Only templates referenced by the bundled entries are vendored. The relative
    depth matches: `../../downloads/X` from `alignment/mechanisms/<role>/<family>/<control>.md` lands in
    alignment/mechanisms/downloads/."""
    dst_dir = os.path.join(spec.mechanisms_dir, "downloads")
    if os.path.isdir(dst_dir):
        shutil.rmtree(dst_dir)  # regenerate fresh so a de-referenced template doesn't linger
    referenced: set[str] = set()
    for f in glob.glob(os.path.join(spec.mechanisms_dir, "**", "*.md"), recursive=True):
        with open(f, encoding="utf-8") as fh:
            referenced.update(re.findall(r"\]\([^)]*downloads/([^)\s#]+)\)", fh.read()))
    n = 0
    for name in sorted(referenced):
        src = os.path.join(ROOT, "downloads", name)
        if os.path.isfile(src):
            _write_copy(src, os.path.join(dst_dir, name))
            n += 1
    return n


def _vendor_explicit_downloads(spec: SkillSpec) -> int:
    """Copy each template named in spec.vendor_downloads into <skill>/templates/ — explicit templates a
    skill ships regardless of what its (possibly empty) bundled entry set references. Regenerated fresh so
    a de-listed template doesn't linger."""
    if not spec.vendor_downloads:
        return 0
    dst_dir = os.path.join(spec.skill_dir, "templates")
    if os.path.isdir(dst_dir):
        shutil.rmtree(dst_dir)
    n = 0
    for name in spec.vendor_downloads:
        src = os.path.join(ROOT, "downloads", name)
        if os.path.isfile(src):
            _write_copy(src, os.path.join(dst_dir, name))
            n += 1
    return n


def _rewrite_outbound_links(spec: SkillSpec) -> None:
    """Repoint links that leave the bundle so none dangles once installed: excluded-product cross-refs
    -> the live public product pages; source-project-internal `docs/` pointers -> plain text (no public
    target). Intra-bundle links (other entries, vendored downloads) are left untouched."""
    targets = glob.glob(os.path.join(spec.mechanisms_dir, "**", "*.md"), recursive=True) + [os.path.join(spec.skill_dir, "principles.md")]
    for f in targets:
        if not os.path.isfile(f):
            continue
        txt = open(f, encoding="utf-8").read()
        txt = re.sub(r"\]\((?:\.\./)+product/([^)\s#]+)\.md\)", lambda m: f"]({PUBLIC}product/{m.group(1)}.html)", txt)
        txt = re.sub(r"\]\((?:\.\./)+product/\)", f"]({PUBLIC}product/)", txt)
        # The construction-kit page is a root catalogue page, not vendored into the bundle (its own links
        # reach the excluded product entries) — repoint entry references to the live published page.
        txt = re.sub(r"\]\((?:\.\./)*constructing-the-gee\.md(#[^)\s]*)?\)",
                     lambda m: f"]({PUBLIC}constructing-the-gee.html{m.group(1) or ''})", txt)
        txt = re.sub(r"\[([^\]]*)\]\(docs/[^)]+\)", r"\1", txt)  # unpublished treatise -> keep text, drop the link
        with open(f, "w", encoding="utf-8") as fh:
            fh.write(txt)


def _bundle(spec: SkillSpec) -> None:
    """Run the full pipeline for one skill. Mirror steps run only when the spec includes roles."""
    print(f"== bundle-skill: {spec.name} ==")
    n = 0
    if spec.include_roles:
        print(f"  roles: {', '.join(spec.include_roles)} (product excluded)")
        n = _mirror_entries(spec)
        _filtered_index(spec)
    _principles(spec)
    d = 0
    if spec.include_roles:
        _reference_readme(spec, n)
        d = _vendor_referenced_downloads(spec)
    _rewrite_outbound_links(spec)
    v = _vendor_explicit_downloads(spec)
    extra = f" + {v} vendored template(s)" if v else ""
    print(f"  -> {os.path.relpath(spec.skill_dir, ROOT)}/ : {n} mechanism entries + principles.md + {d} referenced download(s){extra}")


# ── install / refresh — the adopter half (disjoint upstream-owned / adopter-owned partition) ─────────
MANIFEST_NAME = ".upstream-manifest"  # emitted at the dest skill dir; lists the upstream-owned relpaths
_MANIFEST_HEADER = (
    "# .upstream-manifest — GENERATED by bundle_skill.py (--install / --refresh); do not hand-edit.\n"
    "# Each line below is an UPSTREAM-OWNED file: this tool overwrites it on every `--refresh`, so local\n"
    "# edits to it are lost. Your local additions live in files this tool NEVER touches — a `*.local.md`\n"
    "# sibling (composed APPEND-only over its base at read time) or any file under `local/`. Regen: re-run\n"
    "# `python3 bundle_skill.py --refresh <skill> <dest>`.\n"
)


def _spec_by_name(name: str) -> SkillSpec:
    for s in SPECS:
        if s.name == name:
            return s
    known = ", ".join(s.name for s in SPECS)
    raise SystemExit(f"bundle_skill: unknown managed skill '{name}' (managed skills: {known})")


def _is_adopter_owned(relpath: str) -> bool:
    """The partition predicate. Adopter-owned = a `*.local.md` file OR any path with a directory segment
    named `local`. Everything else is upstream-owned. The two sets are disjoint by construction, so the
    tool never has to merge — it overwrites the upstream set and leaves the adopter set alone."""
    if os.path.basename(relpath).endswith(".local.md"):
        return True
    return "local" in relpath.split(os.sep)[:-1]  # a `local/` DIRECTORY segment (not the basename)


def _upstream_owned_files(src_dir: str) -> list[str]:
    """Every upstream-owned file under the skill's plugin dir, as relpaths. The source's own adopter layer
    (its `*.local.md` / `local/` — e.g. this repo dogfooding as adopter #0) is skipped: it is that repo's
    local tinkering, not content to propagate to a fresh adopter."""
    out: list[str] = []
    for dirpath, _dirnames, filenames in os.walk(src_dir):
        for fn in filenames:
            rel = os.path.relpath(os.path.join(dirpath, fn), src_dir)
            if rel == MANIFEST_NAME or _is_adopter_owned(rel):
                continue
            out.append(rel)
    return sorted(out)


def _read_manifest(dest_dir: str) -> list[str]:
    path = os.path.join(dest_dir, MANIFEST_NAME)
    if not os.path.isfile(path):
        return []
    with open(path, encoding="utf-8") as fh:
        return [ln.strip() for ln in fh if ln.strip() and not ln.startswith("#")]


def _write_manifest(dest_dir: str, upstream: list[str]) -> None:
    _write(os.path.join(dest_dir, MANIFEST_NAME), _MANIFEST_HEADER + "\n".join(upstream) + "\n")


def _existing_adopter_files(dest_dir: str) -> list[str]:
    """Adopter-owned files already present at the dest — reported as PRESERVED (never touched)."""
    if not os.path.isdir(dest_dir):
        return []
    out: list[str] = []
    for dirpath, _dirnames, filenames in os.walk(dest_dir):
        for fn in filenames:
            rel = os.path.relpath(os.path.join(dirpath, fn), dest_dir)
            if rel != MANIFEST_NAME and _is_adopter_owned(rel):
                out.append(rel)
    return sorted(out)


def install_skill(name: str, dest_root: str, *, refresh: bool) -> dict:
    """Install (or refresh) a managed skill from its plugin SSOT into `<dest_root>/<name>/`.

    Overwrites every upstream-owned file; never reads/writes/deletes an adopter-owned file. On refresh,
    a file that was upstream-owned last time (in the old manifest) but is gone upstream now is retired —
    but only if it is still upstream-owned (an adopter-owned path is untouchable). Returns a result dict
    (written / preserved / retired counts + lists) so callers and tests can assert on it.
    """
    spec = _spec_by_name(name)
    src_dir = spec.skill_dir
    if not os.path.isdir(src_dir):
        raise SystemExit(f"bundle_skill: skill source dir not found: {os.path.relpath(src_dir, ROOT)}")
    dest_dir = os.path.join(dest_root, name)

    upstream = _upstream_owned_files(src_dir)
    old_manifest = _read_manifest(dest_dir) if refresh else []
    preserved = _existing_adopter_files(dest_dir)

    # Guard (fail-loud, before any write): the tool must never target an adopter-owned path.
    bad = [r for r in upstream if _is_adopter_owned(r)]
    if bad:
        raise SystemExit(f"bundle_skill: upstream-owned set contains adopter-owned path(s): {bad}")

    for rel in upstream:
        _write_copy(os.path.join(src_dir, rel), os.path.join(dest_dir, rel))

    retired: list[str] = []
    if refresh:
        for rel in old_manifest:
            if rel not in upstream and not _is_adopter_owned(rel):
                p = os.path.join(dest_dir, rel)
                if os.path.isfile(p):
                    os.remove(p)
                    retired.append(rel)

    _write_manifest(dest_dir, upstream)
    return {"skill": name, "dest": dest_dir, "written": upstream, "preserved": preserved, "retired": retired}


def _cmd_install(name: str, dest_root: str, *, refresh: bool) -> int:
    verb = "refresh" if refresh else "install"
    spec = _spec_by_name(name)
    plan = _upstream_owned_files(spec.skill_dir)
    dest_dir = os.path.join(dest_root, name)
    print(f"== bundle-skill {verb}: {name} -> {dest_dir} ==")
    print(f"  plan: overwrite {len(plan)} upstream-owned file(s); preserve every *.local.md / local/** at dest")
    r = install_skill(name, dest_root, refresh=refresh)
    print(f"  -> wrote {len(r['written'])} upstream file(s); "
          f"preserved {len(r['preserved'])} adopter file(s); retired {len(r['retired'])}")
    for rel in r["preserved"]:
        print(f"     preserved (adopter-owned): {rel}")
    for rel in r["retired"]:
        print(f"     retired (gone upstream): {rel}")
    return 0


def _cmd_build() -> int:
    for spec in SPECS:
        if spec.is_generated:
            _bundle(spec)
        else:
            print(f"== bundle-skill: {spec.name} (hand-authored; install-managed, not generated) ==")
    return 0


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Assemble / install the mage plugin's skill bundles.")
    g = ap.add_mutually_exclusive_group()
    g.add_argument("--install", nargs=2, metavar=("SKILL", "DEST"),
                   help="install a managed skill from its plugin SSOT into DEST/<skill>/ (preserves adopter overlays)")
    g.add_argument("--refresh", nargs=2, metavar=("SKILL", "DEST"),
                   help="re-install; overwrite upstream-owned files, retire vanished ones, keep every adopter overlay")
    args = ap.parse_args(argv)
    if args.install:
        return _cmd_install(args.install[0], args.install[1], refresh=False)
    if args.refresh:
        return _cmd_install(args.refresh[0], args.refresh[1], refresh=True)
    return _cmd_build()


if __name__ == "__main__":
    sys.exit(main())