APM-Packages/corentic-package-manager/cpm.py

#!/usr/bin/env python3
"""cpm – Corentic Package Manager.

A CPM package is an agent package consisting of an APM manifest plus a team
configuration (team-profile.yaml). The CPM translates this package into the
native configuration of a target harness. First translation target:
oh-my-opencode-slim. Further targets (oh-my-pi, codex, claude, copilot) are
planned.

Reads a team-profile.yaml, resolves model classes against a local mapping
table and generates a namespaced preset for the target harness.

Commands:
    cpm init        Create user mapping file (optionally fill interactively)
    cpm setup       Guided: read profile, fill mapping, render
    cpm render      Translate team profile into a harness preset and write it
    cpm check       Validate without writing (dry-run)

Examples:
    python3 cpm.py setup  --package examples/package
    python3 cpm.py render --package examples/package --mapping ~/.config/cpm/model-mapping.yaml
    python3 cpm.py check  --package examples/package
"""

from __future__ import annotations

import argparse
import copy
import json
import re
import shutil
import subprocess
import sys
from pathlib import Path

import yaml

# Built-in agents of the first translation target oh-my-opencode-slim.
# Everything else is treated as a custom agent.
BUILTIN_AGENTS = {
    "orchestrator",
    "oracle",
    "librarian",
    "explorer",
    "fixer",
    "designer",
    "council",
    "observer",
}

DEFAULT_MAPPING_PATH = Path.home() / ".config" / "cpm" / "model-mapping.yaml"
DEFAULT_TEAM_FILE = "team-profile.yaml"
DEFAULT_OUTPUT = Path(".opencode") / "oh-my-opencode-slim.json"
PROVENANCE_SUFFIX = ".cpm-provenance.json"

SUPPORTED_SCHEMA = ("corentic.team-profile/v1",)


class CpmError(Exception):
    """Error with a user-readable cause."""


# ---------------------------------------------------------------------------
# Loading


def load_yaml(path: Path) -> dict:
    if not path.exists():
        raise CpmError(f"File not found: {path}")
    try:
        data = yaml.safe_load(path.read_text(encoding="utf-8"))
    except yaml.YAMLError as exc:
        raise CpmError(f"YAML error in {path}: {exc}") from exc
    if not isinstance(data, dict):
        raise CpmError(f"{path} does not contain a YAML mapping")
    return data


def load_team_profile(package_dir: Path, team_file: str | None) -> tuple[dict, Path]:
    candidates = (
        [package_dir / team_file]
        if team_file
        else [package_dir / DEFAULT_TEAM_FILE, package_dir / "team.yaml"]
    )
    for candidate in candidates:
        if candidate.exists():
            profile = load_yaml(candidate)
            schema = profile.get("schema")
            if schema and schema not in SUPPORTED_SCHEMA:
                print(
                    f"Warning: Unknown schema '{schema}'. "
                    f"Supported: {', '.join(SUPPORTED_SCHEMA)}. Continuing."
                )
            if not profile.get("id"):
                raise CpmError(f"{candidate}: missing field 'id'")
            roles = profile.get("roles")
            if not isinstance(roles, list) or not roles:
                raise CpmError(f"{candidate}: no roles defined ('roles')")
            return profile, candidate
    searched = ", ".join(str(c) for c in candidates)
    raise CpmError(f"No team profile found. Searched: {searched}")


def load_mapping(mapping_path: Path | None) -> dict:
    path = mapping_path or DEFAULT_MAPPING_PATH
    if not path.exists():
        raise CpmError(
            f"Model mapping not found: {path}\n"
            "Create the file, e.g.:\n"
            "  model_classes:\n"
            "    fast-research: ollama/qwen3.5:9b\n"
            "    high-reasoning:\n"
            "      model: ollama/qwen3.6:35b-a3b-q4_K_M\n"
            "      variant: thinking"
        )
    data = load_yaml(path)
    classes = data.get("model_classes")
    if not isinstance(classes, dict) or not classes:
        raise CpmError(f"{path}: section 'model_classes' missing or empty")
    return classes


def resolve_model(model_class_entry: object, model_class: str, role_id: str) -> dict:
    """Resolve a mapping entry (string or dict) into agent fields."""
    if isinstance(model_class_entry, str):
        return {"model": model_class_entry}
    if isinstance(model_class_entry, dict):
        entry = {k: v for k, v in model_class_entry.items()}
        if "model" not in entry:
            raise CpmError(
                f"Role '{role_id}': mapping for '{model_class}' has no 'model' field"
            )
        return entry
    raise CpmError(f"Invalid mapping entry for '{model_class}': {model_class_entry!r}")


def sanitize_preset_name(team_id: str) -> str:
    return re.sub(r"[^a-zA-Z0-9_-]+", "-", team_id).strip("-").lower()


# ---------------------------------------------------------------------------
# Translation


def build_preset(profile: dict, mapping: dict) -> tuple[dict, dict, list[str]]:
    """Returns (preset, custom_agents, warnings)."""
    preset: dict = {}
    custom_agents: dict = {}
    warnings: list[str] = []

    for role in profile["roles"]:
        role_id = role.get("id")
        if not role_id:
            raise CpmError("Found a role without 'id'")

        model_class = role.get("model_class")
        if not model_class:
            raise CpmError(f"Role '{role_id}': 'model_class' is missing")
        if model_class not in mapping:
            raise CpmError(
                f"Role '{role_id}': model class '{model_class}' is not mapped.\n"
                f"Add to your mapping:\n"
                f"  {model_class}: <provider/model>"
            )

        agent_fields = resolve_model(mapping[model_class], model_class, role_id)

        capabilities = role.get("capabilities", {}) or {}
        mcps = list(capabilities.get("mcps", []) or [])
        skills = list(capabilities.get("skills", []) or [])
        purpose = str(role.get("purpose", "")).strip()

        runtime_agent = role.get("runtime_agent", role_id)
        if runtime_agent == "custom":
            # Explicitly marked as custom agent -> use the role's own name.
            agent_key = role_id
        else:
            agent_key = runtime_agent

        preset[agent_key] = {
            **agent_fields,
            "mcps": mcps,
            "skills": skills,
        }

        if agent_key in BUILTIN_AGENTS and agent_key != role_id:
            preset[agent_key]["displayName"] = role_id

        if agent_key not in BUILTIN_AGENTS or runtime_agent == "custom":
            if agent_key not in custom_agents:
                prompt = (
                    purpose
                    if purpose
                    else f"Custom agent '{role_id}' from team profile "
                    f"'{profile.get('id', 'unknown')}'."
                )
                custom_agents[agent_key] = {
                    "model": agent_fields["model"],
                    "description": purpose or f"Custom subagent '{role_id}'",
                    "prompt": prompt,
                    # Tell the orchestrator when to delegate to this agent.
                    "orchestratorPrompt": (
                        f"@{agent_key}\n- Role: {purpose}\n"
                        "- Delegate tasks of this role to this agent."
                        if purpose
                        else f"@{agent_key}"
                    ),
                }
            else:
                custom_agents[agent_key]["model"] = agent_fields["model"]

        if not mcps and not skills:
            warnings.append(
                f"Role '{role_id}' ({agent_key}): no MCPs/skills assigned "
                "(purely coordinating?)"
            )

    orchestrator_present = any(k == "orchestrator" for k in preset)
    if not orchestrator_present:
        warnings.append(
            "No 'orchestrator' in the team. Without an orchestrator preset "
            "entry its model stays unchanged."
        )

    return preset, custom_agents, warnings


def merge_into_config(config: dict, preset_name: str, preset: dict,
                      custom_agents: dict) -> dict:
    merged = copy.deepcopy(config)
    presets = merged.setdefault("presets", {})
    existing = presets.get(preset_name)
    if existing:
        print(
            f"Note: preset '{preset_name}' already existed and will be replaced "
            "(other presets remain untouched)."
        )
    presets[preset_name] = preset

    agents = merged.setdefault("agents", {})
    agents.update(custom_agents)
    return merged


# ---------------------------------------------------------------------------
# MCP availability check (best effort)


def strip_jsonc(text: str) -> str:
    text = re.sub(r"/\*.*?\*/", "", text, flags=re.DOTALL)
    text = re.sub(r"(^|\s)//[^\n]*", r"\1", text)
    return text


OPENCODE_CONFIGS = [
    Path.home() / ".config" / "opencode" / "opencode.jsonc",
    Path.home() / ".config" / "opencode" / "opencode.json",
]


def configured_mcps() -> set[str]:
    for path in OPENCODE_CONFIGS:
        if path.exists():
            try:
                data = json.loads(strip_jsonc(path.read_text(encoding="utf-8")))
                return set((data.get("mcp") or {}).keys())
            except (json.JSONDecodeError, OSError):
                continue
    return set()


def available_models() -> list[str]:
    """List all provider/model combinations from opencode.json(c)."""
    models: list[str] = []
    for path in OPENCODE_CONFIGS:
        if not path.exists():
            continue
        try:
            data = json.loads(strip_jsonc(path.read_text(encoding="utf-8")))
        except (json.JSONDecodeError, OSError):
            continue
        for provider, pconf in (data.get("provider") or {}).items():
            for model in (pconf.get("models") or {}):
                models.append(f"{provider}/{model}")
        break
    return sorted(models)


INIT_TEMPLATE = """\
# Local mapping table of the USER.
# Maps abstract model classes to concrete model IDs.
# Formats:
#   class: provider/model
#   class:
#     model: provider/model
#     variant: thinking          # optional
#     temperature: 0.7           # optional

model_classes:
  strong-generalist: PLACEHOLDER
  fast-research: PLACEHOLDER
  strong-writing: PLACEHOLDER
  high-reasoning: PLACEHOLDER
  cheap-reliable: PLACEHOLDER
"""


def check_mcps(preset: dict, warnings: list[str]) -> None:
    available = configured_mcps()
    if not available:
        warnings.append(
            "Could not read opencode.json(c) - MCP check skipped."
        )
        return
    needed: set[str] = set()
    for agent in preset.values():
        needed.update(agent.get("mcps", []) or [])
    missing = sorted(m for m in needed if m not in available)
    if missing:
        warnings.append(
            "MCPs referenced in the team profile but not configured in "
            "opencode.json: " + ", ".join(missing)
        )


# ---------------------------------------------------------------------------
# Commands


def package_output_path(package_dir: Path) -> Path:
    return package_dir / DEFAULT_OUTPUT


def has_apm_dependencies(package_dir: Path) -> bool:
    """True if the package declares an apm.yml with non-empty APM dependencies."""
    manifest = package_dir / "apm.yml"
    if not manifest.exists():
        return False
    try:
        data = load_yaml(manifest)
    except CpmError:
        return False
    deps = (data.get("dependencies") or {}).get("apm") or []
    return bool(deps)


def apm_cli_available() -> bool:
    return shutil.which("apm") is not None


def ensure_apm_install(package_dir: Path, dry_run: bool, interactive: bool) -> None:
    """Make sure the user knows about (and ideally ran) 'apm install'.

    The rendered preset only configures agents; the skills/MCP primitives
    referenced by the team profile are deployed by APM.
    """
    if not has_apm_dependencies(package_dir):
        return

    print(
        "\nAPM dependencies detected: this package references skills/instructions "
        "from other packages that must be installed before the team can use them."
    )
    if dry_run:
        print(f"  Dry run: run 'apm install' in {package_dir} yourself.")
        return

    if apm_cli_available() and interactive and sys.stdin.isatty():
        answer = input("Run 'apm install' now? [y/N] ").strip().lower()
        if answer in ("y", "yes"):
            print()
            result = subprocess.run(["apm", "install"], cwd=package_dir)
            if result.returncode == 0:
                print("\napm install finished successfully.")
            else:
                print(
                    f"\n'apm install' exited with code {result.returncode}. "
                    f"Fix the issue above and re-run it in {package_dir}.",
                    file=sys.stderr,
                )
            return
    else:
        if not apm_cli_available():
            print(
                "  The 'apm' CLI was not found on PATH.\n"
                "  Install it via: pip install apm-cli  (or: brew install microsoft/apm/apm)"
            )
        print(f"  Run manually:\n    cd {package_dir}\n    apm install")


def cmd_render(args: argparse.Namespace) -> int:
    package_dir = Path(args.package).resolve()
    profile, profile_path = load_team_profile(package_dir, args.team_file)
    mapping = load_mapping(Path(args.mapping) if args.mapping else None)

    preset, custom_agents, warnings = build_preset(profile, mapping)

    output = Path(args.output) if args.output else package_dir / DEFAULT_OUTPUT
    if output.exists():
        try:
            config = json.loads(output.read_text(encoding="utf-8"))
        except json.JSONDecodeError as exc:
            raise CpmError(f"{output} is not valid JSON: {exc}")
    else:
        config = {}

    config["$schema"] = (
        "https://unpkg.com/oh-my-opencode-slim@latest/"
        "oh-my-opencode-slim.schema.json"
    )
    merged = merge_into_config(
        config, sanitize_preset_name(str(profile["id"])), preset, custom_agents
    )

    if args.dry_run:
        print(json.dumps(merged, indent=2, ensure_ascii=False))
        _report(warnings, dry_run=True)
        return 0

    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(json.dumps(merged, indent=2, ensure_ascii=False), encoding="utf-8")
    print(f"Written:    {output}")

    provenance = {
        "profile": profile["id"],
        "profileFile": str(profile_path),
        "teamId": profile["id"],
        "description": profile.get("description", ""),
        "presetName": sanitize_preset_name(str(profile["id"])),
        "roles": {
            r.get("id"): {
                "runtimeAgent": r.get("runtime_agent", r.get("id")),
                "modelClass": r.get("model_class"),
                "resolvedModel": mapping[r["model_class"]],
                "mcps": (r.get("capabilities", {}) or {}).get("mcps", []),
                "skills": (r.get("capabilities", {}) or {}).get("skills", []),
            }
            for r in profile["roles"]
        },
    }
    provenance_path = output.with_suffix(PROVENANCE_SUFFIX)
    provenance_path.write_text(
        json.dumps(provenance, indent=2, ensure_ascii=False), encoding="utf-8"
    )
    print(f"Provenance: {provenance_path}")

    _report(warnings)
    steps = [
        "\nNext steps:",
        f"  1. start opencode from {output.parent.parent.resolve()} "
        "(the config lives in its .opencode/ directory)",
    ]
    if has_apm_dependencies(package_dir):
        steps.append(
            "  2. run 'apm install' there first - the team references skills "
            "from APM dependencies that are not deployed yet"
        )
        steps.append(
            f"  3. /preset {sanitize_preset_name(str(profile['id']))}"
        )
        steps.append("  4. reload OpenCode -> team is active")
    else:
        steps.append(
            f"  2. /preset {sanitize_preset_name(str(profile['id']))}"
        )
        steps.append("  3. reload OpenCode -> team is active")
    print("\n".join(steps))
    return 0


def cmd_check(args: argparse.Namespace) -> int:
    args.dry_run = True
    # Dry run: include an existing project config, but never write.
    existing = package_output_path(Path(args.package).resolve())
    args.output = str(existing) if existing.exists() else None
    try:
        cmd_render(args)
    except CpmError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1
    return 0


def _report(warnings: list[str], dry_run: bool = False) -> None:
    if dry_run:
        print("\n--- Dry run: nothing was written ---")
    if warnings:
        print("\nWarnings:")
        for warning in warnings:
            print(f"  ⚠ {warning}")
    else:
        print("\nNo warnings.")


# ---------------------------------------------------------------------------
# Interactive assistants


_MODELS_LISTED = False


def _choose_model(model_class: str, models: list[str]) -> str:
    """Interactively pick a model for a model class."""
    global _MODELS_LISTED
    print(f"\nNo model assigned for '{model_class}'.")
    if models:
        if not _MODELS_LISTED:
            print("Available models (from opencode.json(c)):")
            for i, model in enumerate(models, 1):
                print(f"  {i}. {model}")
            print("  0. Enter model ID manually")
            _MODELS_LISTED = True
        while True:
            choice = input(f"Selection for '{model_class}': ").strip()
            if choice == "0":
                return input("Model ID (provider/model): ").strip()
            if choice.isdigit() and 1 <= int(choice) <= len(models):
                return models[int(choice) - 1]
            print(f"Invalid selection. Enter 1-{len(models)} or 0.")
    else:
        print("No models found in opencode.json(c).")
        return input("Model ID (provider/model): ").strip()


def cmd_init(args: argparse.Namespace) -> int:
    mapping_path = Path(args.mapping) if args.mapping else DEFAULT_MAPPING_PATH
    if mapping_path.exists():
        print(f"Mapping already exists: {mapping_path}")
    else:
        mapping_path.parent.mkdir(parents=True, exist_ok=True)
        mapping_path.write_text(INIT_TEMPLATE, encoding="utf-8")
        print(f"Created: {mapping_path}")

    if args.fill:
        missing = [k for k, v in load_mapping(mapping_path).items()
                   if v == "PLACEHOLDER"]
        models = available_models()
        if missing and not models:
            raise CpmError(
                "No models found in ~/.config/opencode/opencode.json(c). "
                "Enter the model IDs manually or set up opencode first."
            )
        data = load_yaml(mapping_path)
        for model_class in missing:
            data["model_classes"][model_class] = _choose_model(model_class, models)
        mapping_path.write_text(
            yaml.safe_dump(data, sort_keys=False, allow_unicode=True),
            encoding="utf-8",
        )
        remaining = sum(1 for v in data["model_classes"].values()
                        if v == "PLACEHOLDER")
        if remaining:
            print(f"\n{remaining} entr(y/ies) left as PLACEHOLDER - "
                  "please fill in manually.")

    print("\nNext steps:")
    if mapping_path == DEFAULT_MAPPING_PATH:
        print("  Mapping is in place. Validate or render a package, e.g.:")
        print("    cpm.py check --package ./my-cpm-package")
        print("    cpm.py setup --package ./my-cpm-package")
    else:
        print(f"  cpm.py check --package <your-package-dir> --mapping {mapping_path}")
    return 0


def cmd_setup(args: argparse.Namespace) -> int:
    """Guided flow: read profile -> fill mapping -> render."""
    package_dir = Path(args.package).resolve()
    profile, _profile_path = load_team_profile(package_dir, args.team_file)
    mapping_path = Path(args.mapping) if args.mapping else DEFAULT_MAPPING_PATH

    print(f"Team profile: {profile['id']}")
    if profile.get("description"):
        print(f"  {str(profile['description']).strip()}")
    print("Roles:")
    for role in profile["roles"]:
        mcps = (role.get("capabilities", {}) or {}).get("mcps", [])
        skills = (role.get("capabilities", {}) or {}).get("skills", [])
        print(f"  - {role['id']} → {role.get('model_class')} "
              f"(MCPs: {', '.join(mcps) or 'none'}; "
              f"Skills: {', '.join(skills) or 'none'})")

    if not mapping_path.exists():
        print(f"\nMapping missing - creating: {mapping_path}")
        args_init = argparse.Namespace(mapping=str(mapping_path), fill=True)
        cmd_init(args_init)

    classes = load_mapping(mapping_path)
    needed = sorted({r.get("model_class") for r in profile["roles"]
                     if r.get("model_class")} - set(classes))
    models = available_models()
    changed = False
    for model_class in needed:
        chosen = _choose_model(model_class, models)
        classes[model_class] = {"model": chosen, "# selected_by": "user"}
        changed = True
    if changed:
        mapping_path.write_text(
            yaml.safe_dump(classes, sort_keys=False, allow_unicode=True),
            encoding="utf-8",
        )
        print(f"\nMapping updated: {mapping_path}")

    print("\nRendering ...")
    args_render = argparse.Namespace(
        package=str(package_dir),
        team_file=args.team_file,
        mapping=str(mapping_path),
        output=args.output,
        dry_run=args.dry_run,
    )
    result = cmd_render(args_render)

    # Deploy the package primitives (skills/instructions/MCP configs) that
    # the team references. The preset alone only configures the agents.
    ensure_apm_install(
        package_dir,
        dry_run=args.dry_run,
        interactive=not args.dry_run,
    )
    return result


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="cpm",
        description=(
            "Corentic Package Manager. Translates CPM packages (APM manifest + "
            "team-profile.yaml) deterministically into harness-native presets: "
            "team profile + local model mapping -> harness preset."
        ),
    )
    sub = parser.add_subparsers(dest="command", required=True)

    def common(p: argparse.ArgumentParser) -> None:
        p.add_argument(
            "--package",
            default=".",
            help="Path to the CPM package directory (containing team-profile.yaml)",
        )
        p.add_argument(
            "--team-file",
            help="Alternative filename of the team profile (default: team-profile.yaml)",
        )
        p.add_argument(
            "--mapping",
            help=f"Path to the model mapping file (default: {DEFAULT_MAPPING_PATH})",
        )

    p_render = sub.add_parser("render", help="Generate and write the preset")
    common(p_render)
    p_render.add_argument(
        "--output",
        help=f"Target file (default: {DEFAULT_OUTPUT} relative to the package)",
    )
    p_render.add_argument(
        "--dry-run",
        action="store_true",
        help="Print only, write nothing",
    )
    p_render.set_defaults(func=cmd_render)

    p_check = sub.add_parser("check", help="Validate without writing")
    common(p_check)
    p_check.set_defaults(func=cmd_check)

    p_init = sub.add_parser(
        "init",
        help="Create the model mapping file, optionally filling placeholders",
    )
    p_init.add_argument(
        "--mapping",
        help=f"Target path (default: {DEFAULT_MAPPING_PATH})",
    )
    p_init.add_argument(
        "--fill",
        action="store_true",
        help="Interactively choose missing model classes from opencode.json(c) models",
    )
    p_init.set_defaults(func=cmd_init)

    p_setup = sub.add_parser(
        "setup",
        help="Guided: read profile, fill mapping, render",
    )
    common(p_setup)
    p_setup.add_argument(
        "--output",
        help=f"Target file (default: {DEFAULT_OUTPUT} relative to the package)",
    )
    p_setup.add_argument(
        "--dry-run",
        action="store_true",
        help="Print only, write nothing",
    )
    p_setup.set_defaults(func=cmd_setup)

    args = parser.parse_args(argv)
    try:
        return args.func(args)
    except CpmError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1


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