ae6817d86f
The rendered preset only configures agents; the skills/instructions that role allowlists reference are deployed by APM. cpm setup now detects apm.yml dependencies and offers to run 'apm install' after rendering (manual command printed when the CLI is missing or stdin is not a TTY). render/check print it as an explicit next step. Also: - next steps now name the absolute directory to start opencode from (fixes presets appearing 'missing' when opencode runs in the wrong cwd, as hit during first real-world test) - init no longer prints literal <package-dir> placeholders - model list is shown once instead of repeating per class - README: new 'Using the team' section (primitives -> reload -> sanity check -> work via orchestrator), activation writes to the global user config, troubleshooting entry for missing skills
707 lines
24 KiB
Python
707 lines
24 KiB
Python
#!/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())
|