0df9cec833
CPM is an international project; only the blogpost stays German. Translated: cpm.py docstring/messages/help texts, README.md, cpm-setup skill, all team-profile.yaml files (incl. runtime_agent comments), example apm.yml wrappers and the model-mapping template. Also fixes DEFAULT_MAPPING_PATH pointing to ~/.config/apm-team instead of ~/.config/cpm.
614 lines
21 KiB
Python
614 lines
21 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 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 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)
|
||
print(
|
||
"\nNext steps:\n"
|
||
" 1. start opencode (project config gets loaded)\n"
|
||
f" 2. /preset {sanitize_preset_name(str(profile['id']))}\n"
|
||
" 3. reload OpenCode -> team is active"
|
||
)
|
||
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
|
||
|
||
|
||
def _choose_model(model_class: str, models: list[str]) -> str:
|
||
"""Interactively pick a model for a model class."""
|
||
print(f"\nNo model assigned for '{model_class}'.")
|
||
if models:
|
||
print("Available models (from opencode.json(c)):")
|
||
for i, model in enumerate(models, 1):
|
||
print(f" {i}. {model}")
|
||
print(" 0. Enter model ID manually")
|
||
while True:
|
||
choice = input("Selection: ").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("Invalid selection.")
|
||
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 step:")
|
||
print(f" cpm.py check --package <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,
|
||
)
|
||
return cmd_render(args_render)
|
||
|
||
|
||
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())
|