APM-Packages/corentic-package-manager/cpm.py
Tobias J. Endres 2af88b6493 refactor!: rename omos to Corentic Package Manager (CPM)
A CPM package is an agent package consisting of an APM manifest plus a
team configuration (team-profile.yaml). The manager translates such a
package into the native config of a target harness. First target:
oh-my-opencode-slim; oh-my-pi, codex, claude and copilot are planned.

- omos/ -> corentic-package-manager/, cpm.py as CLI entry point
- new commands: cpm init [--fill] and cpm setup (guided flow with
  interactive model selection from opencode.json(c))
- schema field omos_agent -> harness-neutral runtime_agent
- user mapping moved to ~/.config/cpm/model-mapping.yaml
- provenance suffix .cpm-provenance.json
- skill renamed to cpm-setup
2026-08-23 17:26:28 +02:00

617 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""cpm Corentic Package Manager.
Ein CPM-Package ist ein Agent-Package bestehend aus einem APM-Manifest plus
Team-Konfiguration (team-profile.yaml). Der CPM übersetzt dieses Package in
die native Konfiguration eines Ziel-Harnesses. Erstes Übersetzungsziel:
oh-my-opencode-slim. Weitere Ziele (oh-my-pi, codex, claude, copilot) sind
geplant.
Liest ein team-profile.yaml, löst Modellklassen gegen eine lokale
Mapping-Tabelle auf und generiert ein namespacetes Preset für den
Ziel-Harness.
Kommandos:
cpm init Nutzer-Mapping anlegen (optional interaktiv füllen)
cpm setup Geführt: Profil lesen, Mapping auffüllen, rendern
cpm render Team-Profil in Harness-Preset übersetzen und schreiben
cpm check Validieren ohne zu schreiben (Dry-Run)
Beispiele:
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
# Eingebaute Agenten des ersten Übersetzungsziels oh-my-opencode-slim.
# Alles andere wird als Custom Agent behandelt.
BUILTIN_AGENTS = {
"orchestrator",
"oracle",
"librarian",
"explorer",
"fixer",
"designer",
"council",
"observer",
}
DEFAULT_MAPPING_PATH = Path.home() / ".config" / "apm-team" / "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", "acme.team-profile/v1")
class OmosError(Exception):
"""Fehler mit nutzerlesbarer Ursache."""
# ---------------------------------------------------------------------------
# Laden
def load_yaml(path: Path) -> dict:
if not path.exists():
raise OmosError(f"Datei nicht gefunden: {path}")
try:
data = yaml.safe_load(path.read_text(encoding="utf-8"))
except yaml.YAMLError as exc:
raise OmosError(f"YAML-Fehler in {path}: {exc}") from exc
if not isinstance(data, dict):
raise OmosError(f"{path} enthält kein 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"Warnung: Unbekanntes Schema '{schema}'. "
f"Unterstützt: {', '.join(SUPPORTED_SCHEMA)}. Fahre fort."
)
if not profile.get("id"):
raise OmosError(f"{candidate}: Feld 'id' fehlt")
roles = profile.get("roles")
if not isinstance(roles, list) or not roles:
raise OmosError(f"{candidate}: Keine Rollen definiert ('roles')")
return profile, candidate
searched = ", ".join(str(c) for c in candidates)
raise OmosError(f"Kein Team-Profil gefunden. Gesucht: {searched}")
def load_mapping(mapping_path: Path | None) -> dict:
path = mapping_path or DEFAULT_MAPPING_PATH
if not path.exists():
raise OmosError(
f"Modell-Mapping nicht gefunden: {path}\n"
"Lege die Datei an, z. B.:\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 OmosError(f"{path}: Sektion 'model_classes' fehlt oder ist leer")
return classes
def resolve_model(model_class_entry: object, model_class: str, role_id: str) -> dict:
"""Löst einen Mapping-Eintrag (string oder dict) zu Agent-Feldern auf."""
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 OmosError(
f"Rolle '{role_id}': Mapping für '{model_class}' hat kein 'model'-Feld"
)
return entry
raise OmosError(f"Ungültiger Mapping-Eintrag für '{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()
# ---------------------------------------------------------------------------
# Übersetzung
def build_preset(profile: dict, mapping: dict) -> tuple[dict, dict, list[str]]:
"""Erzeugt (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 OmosError("Rolle ohne 'id' gefunden")
model_class = role.get("model_class")
if not model_class:
raise OmosError(f"Rolle '{role_id}': 'model_class' fehlt")
if model_class not in mapping:
raise OmosError(
f"Rolle '{role_id}': Modellklasse '{model_class}' ist nicht gemappt.\n"
f"Ergänze in deinem 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":
# Explizit als Custom Agent markiert → rolleneigener 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}' aus Team-Profil "
f"'{profile.get('id', 'unbekannt')}'."
)
custom_agents[agent_key] = {
"model": agent_fields["model"],
"description": purpose or f"Custom subagent '{role_id}'",
"prompt": prompt,
# Dem Orchestrator sagen, wann er delegieren soll.
"orchestratorPrompt": (
f"@{agent_key}\n- Rolle: {purpose}\n"
"- Delegiere Aufgaben dieser Rolle an diesen Agenten."
if purpose
else f"@{agent_key}"
),
}
else:
custom_agents[agent_key]["model"] = agent_fields["model"]
if not mcps and not skills:
warnings.append(
f"Rolle '{role_id}' ({agent_key}): keine MCPs/Skills zugewiesen "
"(rein koordinierend?)"
)
orchestrator_present = any(k == "orchestrator" for k in preset)
if not orchestrator_present:
warnings.append(
"Kein 'orchestrator' im Team. Ohne Orchestrator-Preset-Eintrag bleibt "
"dessen Modell unverändert."
)
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"Hinweis: Preset '{preset_name}' existierte bereits und wird ersetzt "
"(andere Presets bleiben unberührt)."
)
presets[preset_name] = preset
agents = merged.setdefault("agents", {})
agents.update(custom_agents)
return merged
# ---------------------------------------------------------------------------
# MCP-Verfügbarkeit prüfen (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
def configured_mcps() -> set[str]:
candidates = [
Path.home() / ".config" / "opencode" / "opencode.jsonc",
Path.home() / ".config" / "opencode" / "opencode.json",
]
for path in candidates:
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]:
"""Listet alle provider/model-Kombinationen aus der opencode.json(c)."""
candidates = [
Path.home() / ".config" / "opencode" / "opencode.jsonc",
Path.home() / ".config" / "opencode" / "opencode.json",
]
models: list[str] = []
for path in candidates:
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 = """\
# Lokale Mapping-Tabelle des NUTZERS.
# Ordnet abstrakte Modellklassen konkrete Modell-IDs zu.
# Formate:
# klasse: provider/modell
# klasse:
# model: provider/modell
# 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(
"Konnte opencode.json(c) nicht lesen MCP-Prüfung übersprungen."
)
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 im Team-Profil, aber nicht in opencode.json konfiguriert: "
+ ", ".join(missing)
)
# ---------------------------------------------------------------------------
# Kommandos
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 OmosError(f"{output} ist kein gültiges 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"Geschrieben: {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(
"\nNächste Schritte:\n"
" 1. opencode starten (Projekt-Config wird geladen)\n"
f" 2. /preset {sanitize_preset_name(str(profile['id']))}\n"
" 3. OpenCode neu laden → Team aktiv"
)
return 0
def cmd_check(args: argparse.Namespace) -> int:
args.dry_run = True
# Dry-Run: vorhandene Projekt-Config einbeziehen, aber nie schreiben.
existing = package_output_path(Path(args.package).resolve())
args.output = str(existing) if existing.exists() else None
try:
cmd_render(args)
except OmosError as exc:
print(f"Fehler: {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: Es wurde nichts geschrieben ---")
if warnings:
print("\nWarnungen:")
for warning in warnings:
print(f"{warning}")
else:
print("\nKeine Warnungen.")
# ---------------------------------------------------------------------------
# Interaktive Assistenten
def _choose_model(model_class: str, models: list[str]) -> str:
"""Interaktive Auswahl eines Modells für eine Modellklasse."""
print(f"\nFür '{model_class}' ist kein Modell zugeordnet.")
if models:
print("Verfügbare Modelle (aus opencode.json(c)):")
for i, model in enumerate(models, 1):
print(f" {i}. {model}")
print(" 0. Modell-ID manuell eingeben")
while True:
choice = input("Auswahl: ").strip()
if choice == "0":
return input("Modell-ID (provider/model): ").strip()
if choice.isdigit() and 1 <= int(choice) <= len(models):
return models[int(choice) - 1]
print("Ungültige Auswahl.")
else:
print("Keine Modelle in opencode.json(c) gefunden.")
return input("Modell-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 existiert bereits: {mapping_path}")
else:
mapping_path.parent.mkdir(parents=True, exist_ok=True)
mapping_path.write_text(INIT_TEMPLATE, encoding="utf-8")
print(f"Angelegt: {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 OmosError(
"Keine Modelle in ~/.config/opencode/opencode.json(c) gefunden. "
"Trage die Modell-IDs manuell ein oder richte opencode ein."
)
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} Eintrag/Einträge bleiben PLACEHOLDER "
"bitte manuell ergänzen.")
print("\nNächster Schritt:")
print(f" cpm.py check --package <package-dir> --mapping {mapping_path}")
return 0
def cmd_setup(args: argparse.Namespace) -> int:
"""Geführter Ablauf: Profil lesen → Mapping auffüllen → rendern."""
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-Profil: {profile['id']}")
if profile.get("description"):
print(f" {str(profile['description']).strip()}")
print("Rollen:")
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 'keine'}; "
f"Skills: {', '.join(skills) or 'keine'})")
if not mapping_path.exists():
print(f"\nMapping fehlt lege an: {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 aktualisiert: {mapping_path}")
print("\nRendern …")
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=(
"Adapter zwischen APM-Team-Profilen und oh-my-opencode-slim. "
"Übersetzt deterministisch: Team-Profil + lokales Modell-Mapping "
"-> Harness-Preset."
),
)
sub = parser.add_subparsers(dest="command", required=True)
def common(p: argparse.ArgumentParser) -> None:
p.add_argument(
"--package",
default=".",
help="Pfad zum APM-Package (mit team-profile.yaml)",
)
p.add_argument(
"--team-file",
help="Alternativer Dateiname des Team-Profils (Default: team-profile.yaml)",
)
p.add_argument(
"--mapping",
help=f"Pfad zur Modell-Mapping-Datei (Default: {DEFAULT_MAPPING_PATH})",
)
p_render = sub.add_parser("render", help="Preset generieren und schreiben")
common(p_render)
p_render.add_argument(
"--output",
help=f"Zieldatei (Default: {DEFAULT_OUTPUT} relativ zum Package)",
)
p_render.add_argument(
"--dry-run",
action="store_true",
help="Nur anzeigen, nichts schreiben",
)
p_render.set_defaults(func=cmd_render)
p_check = sub.add_parser("check", help="Validieren ohne zu schreiben")
common(p_check)
p_check.set_defaults(func=cmd_check)
p_init = sub.add_parser(
"init",
help="Modell-Mapping-Datei anlegen und optional Platzhalter füllen",
)
p_init.add_argument(
"--mapping",
help=f"Zielpfad (Default: {DEFAULT_MAPPING_PATH})",
)
p_init.add_argument(
"--fill",
action="store_true",
help="Fehlende Modellklassen interaktiv aus opencode.json(c)-Modellen wählen",
)
p_init.set_defaults(func=cmd_init)
p_setup = sub.add_parser(
"setup",
help="Geführt: Profil lesen, Mapping auffüllen, rendern",
)
common(p_setup)
p_setup.add_argument(
"--output",
help=f"Zieldatei (Default: {DEFAULT_OUTPUT} relativ zum Package)",
)
p_setup.add_argument(
"--dry-run",
action="store_true",
help="Nur anzeigen, nichts schreiben",
)
p_setup.set_defaults(func=cmd_setup)
args = parser.parse_args(argv)
try:
return args.func(args)
except OmosError as exc:
print(f"Fehler: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())