907f43d40b
Deterministic translator between APM team profiles and oh-my-opencode-slim presets. Reads team-profile.yaml, resolves model classes against a user-local mapping table, generates namespaced presets plus provenance. Includes check (dry-run), merge-by-preset-id, custom agent generation from purpose fields, and MCP availability warnings.
427 lines
14 KiB
Python
427 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""omos – Adapter zwischen APM-Team-Profilen und oh-my-opencode-slim.
|
||
|
||
Liest ein team-profile.yaml aus einem APM-Package, löst Modellklassen gegen
|
||
eine lokale Mapping-Tabelle auf und generiert ein namespacetes Preset für
|
||
oh-my-opencode-slim.
|
||
|
||
Kommandos:
|
||
omos render Team-Profil in OMOS-Preset übersetzen und schreiben
|
||
omos check Validieren ohne zu schreiben (Dry-Run)
|
||
|
||
Beispiele:
|
||
python3 omos.py render --package examples/package --mapping ~/.config/apm-team/model-mapping.yaml
|
||
python3 omos.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 OMOS-Agenten. 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 = ".omos-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()
|
||
|
||
omos_agent = role.get("omos_agent", role_id)
|
||
if omos_agent == "custom":
|
||
# Explizit als Custom Agent markiert → rolleneigener Name.
|
||
agent_key = role_id
|
||
else:
|
||
agent_key = omos_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 omos_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 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"): {
|
||
"omosAgent": r.get("omos_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.")
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
parser = argparse.ArgumentParser(
|
||
prog="omos",
|
||
description=(
|
||
"Adapter zwischen APM-Team-Profilen und oh-my-opencode-slim. "
|
||
"Übersetzt deterministisch: Team-Profil + lokales Modell-Mapping "
|
||
"-> OMOS-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)
|
||
|
||
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())
|