Skills › BMAD Method › bmad-forge-idea › resolve_personas.py
resolve_personas.py
365 lines · skills/bmad-forge-idea/scripts/resolve_personas.py at bd70c64
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# ///
5"""Resolve the personas and parties the forge can bring into the room.
7The forge cross-examines witnesses: the installed BMAD agents, plus any
8custom personas and party groups the user has authored for `bmad-party-mode`.
9This surfaces all of them in one shot so the orchestrator never has to ask
10"who's available?" — it just intermixes whoever fits the branch, alongside
11any persona the user names on the fly.
13What it returns (JSON, stdout):
14 * agents — the installed BMAD roster: the default room, always present.
15 * members — extra personas in the pool: guests an installed module's
16 roster offers, and party_members the user defined that
17 aren't already an installed slot.
18 * parties — the named party groups, an installed module's first and then
19 the user's, members resolved to brief entries; open-cast
20 groups (scene names a pool, no roster) are flagged.
21 * default_party — the group id pinned as party-mode's default, if any.
23Discovery is best-effort and never blocks the forge. The installed roster
24comes from `_bmad/scripts/roster.py`, the same source `bmad-party-mode` uses,
25so both skills see the same room; the `[agents]` table of the central config
26is the fallback for a project set up before rosters existed. Custom personas/parties come from
27`bmad-party-mode`'s resolved customization when that skill is found beside
28this one, else from the user's override TOMLs read directly. Anything that
29can't be resolved is simply omitted and flagged, never fatal.
31Stdlib only (Python 3.11+ for tomllib).
33 resolve_personas.py --project-root P --skill S
34"""
36import argparse
37import json
38import subprocess
39import sys
40from pathlib import Path
42try:
43 import tomllib
44except ImportError: # pragma: no cover - guarded for <3.11
45 sys.stderr.write("error: Python 3.11+ is required (stdlib `tomllib`).\n")
46 sys.exit(3)
48PARTY_SKILL = "bmad-party-mode"
51# Why the last _run_json call failed; tests stub _run_json with one argument.
52_last_error = ""
55def _run_json(cmd):
56 """Run a resolver script and parse its JSON stdout. None on any failure."""
57 global _last_error
58 _last_error = ""
59 try:
60 out = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", timeout=60)
61 except (OSError, subprocess.SubprocessError) as exc:
62 _last_error = str(exc)
63 return None
64 if out.returncode != 0 or not out.stdout.strip():
65 _last_error = out.stderr.strip() or f"exit {out.returncode}, no output"
66 return None
67 try:
68 return json.loads(out.stdout)
69 except json.JSONDecodeError as exc:
70 _last_error = f"resolver printed invalid JSON: {exc}"
71 return None
74def _load_toml(path: Path):
75 if not path.exists():
76 return {}
77 try:
78 with path.open("rb") as f:
79 data = tomllib.load(f)
80 return data if isinstance(data, dict) else {}
81 except (OSError, tomllib.TOMLDecodeError):
82 return {}
85def load_roster(project_root: Path, skill_root: Path):
86 """(agents, guests, groups, resolved_ok) from the skills installed beside this one.
88 agents are {code: entry} for the default room. guests are the other
89 roster members, available by name or through a group, as in party mode.
90 """
91 scripts = project_root / "_bmad" / "scripts"
92 data = _run_json(
93 [sys.executable, str(scripts / "roster.py"), "--skill", str(skill_root), "--project-root", str(project_root)]
94 )
95 if data is not None:
96 agents = data.get("agents", {})
97 members = data.get("members", {})
98 groups = data.get("groups", [])
99 agents = agents if isinstance(agents, dict) else {}
100 guests = {
101 code: member
102 for code, member in (members if isinstance(members, dict) else {}).items()
103 if code not in agents and isinstance(member, dict)
104 }
105 return agents, guests, groups if isinstance(groups, list) else [], True
106 agents, resolved = load_config_agents(project_root)
107 return agents, {}, [], resolved
110def load_config_agents(project_root: Path):
111 """The central config's [agents] table as {code: entry}. (dict, resolved_ok).
113 The core resolver may emit agents as a dict keyed by code or as an array
114 of tables (depending on how the layers merged); normalize both to a dict.
115 """
116 script = project_root / "_bmad" / "scripts" / "resolve_config.py"
117 data = _run_json([sys.executable, str(script), "--project-root", str(project_root), "--key", "agents"])
118 if data is None:
119 return {}, False
120 agents = data.get("agents", {}) or {}
121 if isinstance(agents, list):
122 agents = {a["code"]: a for a in agents if isinstance(a, dict) and a.get("code")}
123 elif not isinstance(agents, dict):
124 agents = {}
125 return agents, True
128def merge_groups(roster_groups: list, custom_groups: list) -> list:
129 """Roster groups first, then the user's; a custom group replaces a roster group with its id."""
130 merged = {g["id"]: g for g in roster_groups if isinstance(g, dict) and g.get("id")}
131 for g in custom_groups if isinstance(custom_groups, list) else []:
132 if isinstance(g, dict) and g.get("id"):
133 merged[g["id"]] = g
134 return list(merged.values())
137def find_party_skill(project_root: Path, skill_root: Path):
138 """Locate the installed bmad-party-mode skill dir, or None.
140 Skills install as siblings, so the party skill is almost always next to
141 this one. A couple of common install roots cover the rest.
142 """
143 candidates = [
144 skill_root.parent / PARTY_SKILL,
145 project_root / ".claude" / "skills" / PARTY_SKILL,
146 project_root / "_bmad" / "skills" / PARTY_SKILL,
147 ]
148 for c in candidates:
149 if (c / "customize.toml").exists():
150 return c
151 return None
154def load_party_workflow(project_root: Path, party_skill: Path):
155 """Merged [workflow] table for bmad-party-mode (base + user overrides)."""
156 resolver = project_root / "_bmad" / "scripts" / "resolve_customization.py"
157 data = _run_json(
158 [
159 sys.executable,
160 str(resolver),
161 "--skill",
162 str(party_skill),
163 "--project-root",
164 str(project_root),
165 "--key",
166 "workflow",
167 ]
168 )
169 if data is not None and isinstance(data.get("workflow"), dict):
170 return data["workflow"]
171 _warn(f"party customization override not applied, using the shipped party: {_last_error or 'no workflow table'}")
172 # Fallback: base customize.toml directly, no override merge.
173 wf = _load_toml(party_skill / "customize.toml").get("workflow", {})
174 return wf if isinstance(wf, dict) else {}
177def load_party_overrides(project_root: Path):
178 """Custom personas/parties when party-mode itself isn't installed.
180 Reads only the user's override TOMLs (team then personal, personal wins on
181 scalars). No base roster exists in this path, so a shallow merge is enough.
182 """
183 custom = project_root / "_bmad" / "custom"
184 team = _load_toml(custom / f"{PARTY_SKILL}.toml").get("workflow", {})
185 user = _load_toml(custom / f"{PARTY_SKILL}.user.toml").get("workflow", {})
186 team = team if isinstance(team, dict) else {}
187 user = user if isinstance(user, dict) else {}
188 merged = dict(team)
189 for key, val in user.items():
190 if isinstance(val, list) and isinstance(merged.get(key), list):
191 merged[key] = merged[key] + val
192 else:
193 merged[key] = val
194 return merged
197def _warn(message: str):
198 sys.stderr.write(f"warning: {message}\n")
201def _bad_member(code, name) -> bool:
202 """True, with a warning, when a member's code or name is not a string."""
203 if isinstance(code, str) and (name is None or isinstance(name, str)):
204 return False
205 _warn(f"persona {code!r} left out: code and name must be strings")
206 return True
209def _alias(code: str) -> str:
210 """Short alias for an installed agent code: bmad-agent-analyst -> analyst."""
211 for prefix in ("bmad-agent-", "bmad-"):
212 if code.startswith(prefix):
213 return code[len(prefix) :]
214 return code
217def build_pool(agents: dict, party_members: list, guests: dict | None = None):
218 """One pool keyed by code; custom members override matching installed slots.
220 Returns (pool, index, installed_codes, custom_codes):
221 * installed_codes — the default room (installed agents, overrides applied
222 in place); custom-only additions stay in the pool but don't crowd it.
223 * custom_codes — pure-custom personas (no installed slot), the extra
224 faces the forge can summon by name or via a party group.
225 """
226 pool, index, installed_codes, custom_codes = {}, {}, [], []
228 def register(code, entry):
229 pool[code] = entry
230 index[code] = code
231 index[code.lower()] = code
232 index[_alias(code).lower()] = code
233 name = entry.get("name")
234 if name:
235 key = name.lower()
236 # A custom rename must not hijack another agent's name lookup.
237 if index.get(key, code) == code:
238 index[key] = code
240 for code, info in (agents or {}).items():
241 if _bad_member(code, info.get("name")):
242 continue
243 register(
244 code,
245 {
246 "code": code,
247 "name": info.get("name", code),
248 "icon": info.get("icon", ""),
249 "title": info.get("title", ""),
250 "description": info.get("description", ""),
251 "persona": info.get("persona", ""),
252 "source": "installed",
253 },
254 )
255 installed_codes.append(code)
257 for code, info in (guests or {}).items():
258 if _bad_member(code, info.get("name")):
259 continue
260 entry = {"code": code, "source": "roster"}
261 for field in ("name", "icon", "title", "persona", "capabilities", "model"):
262 if info.get(field):
263 entry[field] = info[field]
264 entry.setdefault("name", code)
265 register(code, entry)
266 custom_codes.append(code)
268 for m in party_members if isinstance(party_members, list) else []:
269 if not isinstance(m, dict):
270 continue
271 code = m.get("code")
272 if code is None or code == "" or _bad_member(code, m.get("name")):
273 continue
274 canonical = index.get(code) or index.get(code.lower()) or code
275 was_installed = canonical in pool
276 # Start from the installed entry so fields the override omits
277 # (icon, title, description) survive.
278 entry = dict(pool.get(canonical, {}))
279 entry.update({"code": canonical, "source": "custom"})
280 for field in ("name", "icon", "title", "persona", "capabilities", "model"):
281 if m.get(field) is not None:
282 entry[field] = m[field]
283 entry.setdefault("name", canonical)
284 register(canonical, entry)
285 if not was_installed:
286 custom_codes.append(canonical)
288 return pool, index, installed_codes, custom_codes
291def _brief(entry):
292 """The slim card the orchestrator needs to cast a persona."""
293 out = {k: entry[k] for k in ("code", "name", "icon", "title", "source") if entry.get(k)}
294 for k in ("description", "persona", "capabilities", "model"):
295 if entry.get(k):
296 out[k] = entry[k]
297 return out
300def resolve_parties(groups, pool, index):
301 out = []
302 for g in groups or []:
303 if not isinstance(g, dict) or not g.get("id"):
304 continue
305 raw = g.get("members", []) or []
306 members = []
307 for t in raw:
308 key = t if isinstance(t, str) else str(t)
309 code = index.get(key) or index.get(key.lower())
310 if code in pool:
311 members.append(_brief(pool[code]))
312 party = {"id": g["id"], "name": g.get("name", g["id"]), "members": members}
313 if g.get("scene"):
314 party["scene"] = g["scene"]
315 if not raw:
316 party["open_cast"] = True
317 out.append(party)
318 return out
321def main():
322 ap = argparse.ArgumentParser(description="Resolve forge personas and parties.")
323 ap.add_argument("--project-root", required=True)
324 ap.add_argument("--skill", required=True, help="Path to the bmad-forge-idea skill dir")
325 args = ap.parse_args()
327 project_root = Path(args.project_root).resolve()
328 skill_root = Path(args.skill).resolve()
330 agents, guests, roster_groups, agents_ok = load_roster(project_root, skill_root)
332 party_skill = find_party_skill(project_root, skill_root)
333 if party_skill is not None:
334 workflow = load_party_workflow(project_root, party_skill)
335 else:
336 workflow = load_party_overrides(project_root)
338 pool, index, installed_codes, custom_codes = build_pool(agents, workflow.get("party_members", []), guests)
339 parties = resolve_parties(merge_groups(roster_groups, workflow.get("party_groups", [])), pool, index)
341 _emit(
342 {
343 "agents": [_brief(pool[c]) for c in installed_codes],
344 "members": [_brief(pool[c]) for c in custom_codes],
345 "parties": parties,
346 "default_party": workflow.get("default_party", "") or "",
347 "party_mode_found": party_skill is not None,
348 "agents_resolved": agents_ok,
349 }
350 )
353def _emit(obj):
354 reconfigure = getattr(sys.stdout, "reconfigure", None)
355 if reconfigure is not None:
356 reconfigure(encoding="utf-8")
357 sys.stdout.write(json.dumps(obj, indent=2, ensure_ascii=False) + "\n")
360if __name__ == "__main__":
361 if sys.platform == "win32":
362 # Piped output on Windows defaults to a legacy code page, not UTF-8.
363 sys.stdout.reconfigure(encoding="utf-8")
364 sys.stderr.reconfigure(encoding="utf-8")
365 main()