Skills › BMAD Method › bmad-party-mode › resolve_party.py
resolve_party.py
389 lines · skills/bmad-party-mode/scripts/resolve_party.py at bd70c64
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# ///
5"""Resolve the party-mode roster, lazily.
7Merges the roster the installed skills offer (agents, guests and groups, read
8by `_bmad/scripts/roster.py`) with the user's custom `party_members` and
9`party_groups` into one collective, then projects only what the moment needs:
11 * default (no flag) — the active roster to load on entry: the
12 `default_party` group if one is configured, else the whole collective.
13 Other groups come back as names only, so nothing you aren't using is
14 loaded into the party.
15 * --list-groups — just id + name + size for every configured group. The
16 cheap menu for "which room?", with no member detail.
17 * --party <id> — full member detail for one chosen group, on demand
18 (e.g. when the user switches rooms). Unknown id returns the available
19 names instead of an error wall.
21The merge is deterministic (a keyed union; a custom member whose code
22matches an installed agent overrides it), so the orchestrator consumes a
23resolved roster instead of re-deriving it every session.
25Stdlib only (Python 3.11+ for tomllib). Shells out to the project's roster.py
26and resolve_customization.py. An install whose `_bmad/scripts` predates
27roster.py falls back to the `[agents]` table resolve_config.py returns, and a
28missing customization resolver falls back to reading customize.toml directly.
30 resolve_party.py --project-root P --skill S
31 resolve_party.py --project-root P --skill S --list-groups
32 resolve_party.py --project-root P --skill S --party writers-room (alias: --group)
33"""
35import argparse
36import json
37import subprocess
38import sys
39from pathlib import Path
41try:
42 import tomllib
43except ImportError: # pragma: no cover - guarded for <3.11
44 sys.stderr.write("error: Python 3.11+ is required (stdlib `tomllib`).\n")
45 sys.exit(3)
48# Why the last _run_json call failed; tests stub _run_json with one argument.
49_last_error = ""
52def _run_json(cmd):
53 """Run a resolver script and parse its JSON stdout. None on any failure."""
54 global _last_error
55 _last_error = ""
56 try:
57 out = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", timeout=60)
58 except (OSError, subprocess.SubprocessError) as exc:
59 _last_error = str(exc)
60 return None
61 if out.returncode != 0 or not out.stdout.strip():
62 _last_error = out.stderr.strip() or f"exit {out.returncode}, no output"
63 return None
64 try:
65 return json.loads(out.stdout)
66 except json.JSONDecodeError as exc:
67 _last_error = f"resolver printed invalid JSON: {exc}"
68 return None
71def load_roster(project_root: Path, skill_root: Path):
72 """(agents, guests, groups, resolved, problems) from the skills installed beside this one.
74 agents are {code: entry} for the default room; guests are roster members
75 that are not installed agents, available to groups only. problems are what
76 roster.py could not use, so a missing agent or room is explained.
77 """
78 scripts = project_root / "_bmad" / "scripts"
79 data = _run_json(
80 [sys.executable, str(scripts / "roster.py"), "--skill", str(skill_root), "--project-root", str(project_root)]
81 )
82 if data is not None:
83 agents = data.get("agents", {}) or {}
84 guests = {code: m for code, m in (data.get("members", {}) or {}).items() if code not in agents}
85 problems = [p.get("problem", "") for p in data.get("problems", []) or [] if isinstance(p, dict)]
86 return agents, guests, data.get("groups", []) or [], True, [p for p in problems if p]
87 data = _run_json(
88 [sys.executable, str(scripts / "resolve_config.py"), "--project-root", str(project_root), "--key", "agents"]
89 )
90 if data is None:
91 return {}, {}, [], False, []
92 return data.get("agents", {}) or {}, {}, [], True, []
95def merge_groups(roster_groups: list, custom_groups: list) -> list:
96 """Roster groups first, then the user's; a custom group replaces a roster group with its id."""
97 merged = {g["id"]: g for g in roster_groups if isinstance(g, dict) and g.get("id")}
98 for g in custom_groups or []:
99 if isinstance(g, dict) and g.get("id"):
100 merged[g["id"]] = g
101 return list(merged.values())
104def load_workflow(project_root: Path, skill_root: Path):
105 """Merged [workflow] table. Falls back to the skill's base customize.toml."""
106 script = project_root / "_bmad" / "scripts" / "resolve_customization.py"
107 data = _run_json(
108 [
109 sys.executable,
110 str(script),
111 "--skill",
112 str(skill_root),
113 "--project-root",
114 str(project_root),
115 "--key",
116 "workflow",
117 ]
118 )
119 if data is not None and "workflow" in data:
120 return data["workflow"]
121 _warn(f"party customization override not applied, using the shipped party: {_last_error or 'no workflow table'}")
122 # Fallback: read the skill's base customize.toml directly (no override merge).
123 toml_path = skill_root / "customize.toml"
124 if toml_path.exists():
125 try:
126 with toml_path.open("rb") as f:
127 return tomllib.load(f).get("workflow", {})
128 except (OSError, tomllib.TOMLDecodeError):
129 pass
130 return {}
133def _warn(message: str):
134 sys.stderr.write(f"warning: {message}\n")
137def _bad_member(code, name) -> bool:
138 """True, with a warning, when a member's code or name is not a string."""
139 if isinstance(code, str) and (name is None or isinstance(name, str)):
140 return False
141 _warn(f"party member {code!r} left out: code and name must be strings")
142 return True
145def _alias(code: str) -> str:
146 """Short alias for an installed agent code: bmad-agent-analyst -> analyst."""
147 if "-agent-" in code:
148 return code.split("-agent-", 1)[1]
149 for prefix in ("bmad-agent-", "bmad-"):
150 if code.startswith(prefix):
151 return code[len(prefix) :]
152 return code
155def build_collective(agents: dict, party_members: list, guests: dict | None = None):
156 """One pool keyed by code. Custom members override matching installed agents.
158 `guests` are roster members that are not installed agents: a module's extra
159 personas, or an agent whose skill is absent. They join the pool so a group
160 can seat them, and never the default room.
162 Returns (collective, index, installed_codes):
163 * collective — every member (installed + custom), the pool groups draw
164 from and the orchestrator can summon by name.
165 * index — maps every resolvable token (code, prefix-stripped alias,
166 lower-cased name) to a canonical code.
167 * installed_codes — the codes occupying an installed-agent slot, in
168 order. This is the DEFAULT room: installed agents (with any custom
169 override applied in place), and NOT the pure-custom additions. So
170 shipping or defining custom members grows the pool without crowding
171 the default party.
172 """
173 collective = {}
174 index = {}
175 installed_codes = []
176 alias_owner = {}
178 def register(code, entry):
179 collective[code] = entry
180 index[code] = code
181 index[code.lower()] = code
182 # A short alias two codes claim resolves to neither; the full code still works.
183 alias = _alias(code).lower()
184 owner = alias_owner.setdefault(alias, code)
185 if owner == code:
186 index.setdefault(alias, code)
187 elif owner is not None:
188 alias_owner[alias] = None
189 if index.get(alias) == owner and alias != owner.lower():
190 del index[alias]
191 name = entry.get("name")
192 if name:
193 index[name.lower()] = code
195 for code, info in agents.items():
196 if _bad_member(code, info.get("name")):
197 continue
198 entry = {
199 "code": code,
200 "name": info.get("name", code),
201 "icon": info.get("icon", ""),
202 "title": info.get("title", ""),
203 "module": info.get("module", ""),
204 "source": "installed",
205 }
206 # Installs from before rosters recorded the persona as `description`.
207 persona = info.get("persona") or info.get("description")
208 if persona:
209 entry["persona"] = persona
210 for field in ("capabilities", "model"):
211 if info.get(field):
212 entry[field] = info[field]
213 register(code, entry)
214 installed_codes.append(code)
216 for code, info in (guests or {}).items():
217 if _bad_member(code, info.get("name")):
218 continue
219 entry = {"code": code, "source": "roster"}
220 for field in ("name", "icon", "title", "persona", "capabilities", "model", "module", "skill", "install"):
221 if info.get(field):
222 entry[field] = info[field]
223 entry.setdefault("name", code)
224 if info.get("installed") is False:
225 entry["installed"] = False
226 register(code, entry)
228 for m in party_members if isinstance(party_members, list) else []:
229 if not isinstance(m, dict):
230 continue
231 code = m.get("code")
232 if code is None or code == "" or _bad_member(code, m.get("name")):
233 continue
234 # A custom member overrides an installed agent it matches by code/alias/name.
235 canonical = index.get(code) or index.get(code.lower()) or code
236 # Start from the installed entry so fields the override omits
237 # (icon, title, persona, module) survive.
238 entry = dict(collective.get(canonical, {}))
239 entry.update({"code": canonical, "source": "custom"})
240 for field in ("name", "icon", "title", "persona", "capabilities", "model"):
241 if m.get(field) is not None:
242 entry[field] = m[field]
243 entry.setdefault("name", canonical)
244 register(canonical, entry)
245 # An override keeps the installed slot; a brand-new custom does not join it.
247 return collective, index, installed_codes
250def resolve_members(member_tokens, collective, index):
251 """(resolved entries in listed order, unresolved tokens)."""
252 resolved, unresolved = [], []
253 for token in member_tokens or []:
254 if not isinstance(token, str):
255 unresolved.append(token) # malformed config value — never a key lookup
256 continue
257 code = index.get(token) or index.get(token.lower())
258 if code and code in collective:
259 resolved.append(collective[code])
260 else:
261 unresolved.append(token)
262 return resolved, unresolved
265def group_menu(groups):
266 """Names only — the cheap menu. Open-cast groups (no roster) are flagged."""
267 out = []
268 for g in groups or []:
269 if not isinstance(g, dict) or not g.get("id"):
270 continue
271 members = g.get("members", []) or []
272 entry = {"id": g["id"], "name": g.get("name", g["id"]), "member_count": len(members)}
273 if not members:
274 entry["open_cast"] = True
275 out.append(entry)
276 return out
279def find_group(groups, group_id):
280 for g in groups or []:
281 if isinstance(g, dict) and g.get("id") == group_id:
282 return g
283 return None
286def group_detail(g, collective, index):
287 """Full detail for one group: resolved members + the optional scene.
289 `scene` is a freeform line the orchestrator plays — setting, what's
290 happening, room dynamics, in-the-moment character notes. Surfaced only
291 here (when a group is the active/chosen roster), never in the menu.
293 `members` is optional. With none, the group is open-cast: `open_cast`
294 is flagged and the scene describes the pool the orchestrator casts from
295 on the fly (e.g. "figures from the Star Wars Rebels universe"). A few
296 listed members anchor the room; the scene can still invite more.
297 """
298 raw_members = g.get("members", []) or []
299 members, unresolved = resolve_members(raw_members, collective, index)
300 detail = {
301 "active": g["id"],
302 "name": g.get("name", g["id"]),
303 "members": members,
304 "unresolved": unresolved,
305 "memory_enabled": bool(g.get("memory", False)),
306 }
307 if g.get("scene"):
308 detail["scene"] = g["scene"]
309 if not raw_members:
310 detail["open_cast"] = True
311 return detail
314def build_parser():
315 ap = argparse.ArgumentParser(description="Resolve the party-mode roster, lazily.")
316 ap.add_argument("--project-root", required=True)
317 ap.add_argument("--skill", required=True, help="Path to the bmad-party-mode skill dir")
318 ap.add_argument("--party", "--group", dest="party", help="Resolve full detail for this group id")
319 ap.add_argument("--list-groups", action="store_true", help="Group names only")
320 return ap
323def main():
324 args = build_parser().parse_args()
326 project_root = Path(args.project_root).resolve()
327 skill_root = Path(args.skill).resolve()
329 workflow = load_workflow(project_root, skill_root)
330 agents, guests, roster_groups, agents_ok, roster_problems = load_roster(project_root, skill_root)
331 groups = merge_groups(roster_groups, workflow.get("party_groups", []) or [])
332 default_party = workflow.get("default_party", "") or ""
333 party_mode = workflow.get("party_mode", "session") or "session"
334 # The global party_memory flag governs only the DEFAULT installed-agent room;
335 # a named group carries its own `memory` flag (resolved in group_detail).
336 party_memory = bool(workflow.get("party_memory", True))
338 if args.list_groups:
339 _emit(
340 {
341 "party_mode": party_mode,
342 "default_party": default_party,
343 "groups": group_menu(groups),
344 }
345 )
346 return
348 collective, index, installed_codes = build_collective(agents, workflow.get("party_members", []), guests)
350 if args.party:
351 g = find_group(groups, args.party)
352 if g is None:
353 _emit({"error": "unknown_group", "requested": args.party, "available": group_menu(groups)})
354 return
355 detail = {**group_detail(g, collective, index), "party_mode": party_mode}
356 if roster_problems:
357 detail["roster_problems"] = roster_problems
358 _emit(detail)
359 return
361 # Default: the active roster to load on entry.
362 result = {"party_mode": party_mode, "groups": group_menu(groups), "installed_agents_resolved": agents_ok}
363 if roster_problems:
364 result["roster_problems"] = roster_problems
365 g = find_group(groups, default_party) if default_party else None
366 if g is not None:
367 result.update(group_detail(g, collective, index))
368 else:
369 # No default group: the installed agents (custom additions stay in the
370 # pool but don't crowd the default room), exactly like a plain install.
371 result.update(
372 {"active": "installed", "members": [collective[c] for c in installed_codes], "memory_enabled": party_memory}
373 )
374 _emit(result)
377def _emit(obj):
378 reconfigure = getattr(sys.stdout, "reconfigure", None)
379 if reconfigure is not None:
380 reconfigure(encoding="utf-8")
381 sys.stdout.write(json.dumps(obj, indent=2, ensure_ascii=False) + "\n")
384if __name__ == "__main__":
385 if sys.platform == "win32":
386 # Piped output on Windows defaults to a legacy code page, not UTF-8.
387 sys.stdout.reconfigure(encoding="utf-8")
388 sys.stderr.reconfigure(encoding="utf-8")
389 main()