428 lines
14 KiB
Python
428 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())
|