Skills › BMAD Method › bmad › render_skill.py
render_skill.py
757 lines · skills/bmad/scripts/render_skill.py at bd70c64
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# dependencies = ["jinja2>=3.1"]
5# ///
6"""Render a skill's Markdown sources into an immutable project snapshot."""
8from __future__ import annotations
10import argparse
11import hashlib
12import json
13import os
14import re
15import shutil
16import sys
17import tomllib
18from collections.abc import Callable
19from datetime import date, time
20from pathlib import Path
21from typing import Any
23import jinja2
25# Installed scripts are consumer files, not a location for interpreter caches.
26sys.dont_write_bytecode = True
28from config_utils import ( # noqa: E402
29 ConfigError,
30 load_central_config,
31 load_customization,
32 load_toml,
33 structural_merge,
34)
37class RenderError(ValueError):
38 """Raised when rendering cannot safely publish a snapshot."""
41class _ArgumentParser(argparse.ArgumentParser):
42 def error(self, message: str) -> None:
43 raise RenderError(message)
46_PARAMETER = r"[A-Za-z0-9_-]+(?:\.[A-Za-z0-9_-]+)*"
49def _hash_bytes(content: bytes) -> str:
50 return hashlib.sha256(content).hexdigest()
53def _canonical_json(value: Any) -> bytes:
54 return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=_json_scalar).encode(
55 "utf-8"
56 )
59def _json_scalar(value: Any) -> str:
60 if isinstance(value, (date, time)):
61 return value.isoformat()
62 raise TypeError(f"unsupported JSON value: {type(value).__name__}")
65def _toml_literal(text: str, label: str) -> Any:
66 try:
67 parsed = tomllib.loads(f"value = {text}")
68 except tomllib.TOMLDecodeError as error:
69 raise RenderError(f"invalid TOML value for {label}: {error}") from error
70 if set(parsed) != {"value"}:
71 raise RenderError(f"{label} must contain a single TOML value")
72 return parsed["value"]
75def _invocation_customization(
76 defaults: dict[str, Any], overrides: Path | None, assignments: list[str]
77) -> tuple[dict[str, Any], dict[str, Any]]:
78 file_layer = load_toml(overrides, required=True) if overrides is not None else {}
79 command_layer: dict[str, Any] = {}
80 assigned: list[str] = []
81 for assignment in assignments:
82 path, separator, raw = assignment.partition("=")
83 if not separator or re.fullmatch(_PARAMETER, path) is None:
84 raise RenderError(f"invalid --set assignment {assignment!r}; expected bare dotted key=value")
85 # A repeated or overlapping path is a caller mistake, not a precedence rule.
86 for earlier in assigned:
87 if path == earlier or path.startswith(f"{earlier}.") or earlier.startswith(f"{path}."):
88 raise RenderError(f"--set `{path}` conflicts with earlier --set `{earlier}`")
89 assigned.append(path)
90 default = _lookup(defaults, path, "customization parameter")
91 value = (
92 raw if isinstance(default, str) and not raw.lstrip().startswith(('"', "'")) else _toml_literal(raw, path)
93 )
94 target = command_layer
95 parts = path.split(".")
96 for part in parts[:-1]:
97 target = target.setdefault(part, {})
98 target[parts[-1]] = value
99 return file_layer, command_layer
102def _leaf_paths(table: dict[str, Any], prefix: str = "") -> set[str]:
103 leaves: set[str] = set()
104 for key, value in table.items():
105 path = f"{prefix}{key}"
106 if isinstance(value, dict):
107 leaves |= _leaf_paths(value, f"{path}.")
108 else:
109 leaves.add(path)
110 return leaves
113def _declares(defaults: dict[str, Any], path: str) -> bool:
114 node: Any = defaults
115 for part in path.split("."):
116 if not isinstance(node, dict) or part not in node:
117 return False
118 node = node[part]
119 return True
122def _check_persistent_layers(project_root: Path, skill_dir: Path, defaults: dict[str, Any] | None) -> None:
123 """A persistent override may only set keys the skill declares; a stale or misspelled key halts."""
124 custom_dir = project_root / "_bmad" / "custom"
125 for layer in (custom_dir / f"{skill_dir.name}.toml", custom_dir / f"{skill_dir.name}.user.toml"):
126 undeclared = sorted(path for path in _leaf_paths(load_toml(layer)) if not _declares(defaults or {}, path))
127 if undeclared:
128 raise RenderError(f"{layer} sets keys {skill_dir.name} does not declare: {', '.join(undeclared)}")
131def _lookup(data: dict[str, Any], dotted_path: str, label: str) -> Any:
132 current: Any = data
133 for part in dotted_path.split("."):
134 if not isinstance(current, dict) or part not in current:
135 raise RenderError(f"missing {label} `{dotted_path}`")
136 current = current[part]
137 return current
140def _require_string(value: Any, label: str, *, allow_empty: bool = False) -> str:
141 if not isinstance(value, str):
142 raise RenderError(f"{label} must be a string, got {type(value).__name__}")
143 if not allow_empty and not value.strip():
144 raise RenderError(f"{label} must not be empty")
145 return value
148def _require_string_list(value: Any, label: str) -> list[str]:
149 if not isinstance(value, list):
150 raise RenderError(f"{label} must be a list, got {type(value).__name__}")
151 result = []
152 for index, item in enumerate(value):
153 result.append(_require_string(item, f"{label}[{index}]"))
154 return result
157def _require_review_layers(value: Any, label: str) -> list[dict[str, str]]:
158 if not isinstance(value, list):
159 raise RenderError(f"{label} must be a list of tables")
160 result: list[dict[str, str]] = []
161 seen: set[str] = set()
162 for index, item in enumerate(value):
163 item_label = f"{label}[{index}]"
164 if not isinstance(item, dict):
165 raise RenderError(f"{item_label} must be a table")
166 identifier = _require_string(item.get("id"), f"{item_label}.id")
167 if identifier in seen:
168 raise RenderError(f"duplicate review layer id `{identifier}`")
169 seen.add(identifier)
170 layer = {
171 "id": identifier,
172 "name": _require_string(item.get("name", identifier), f"{item_label}.name"),
173 "instruction": _require_string(item.get("instruction"), f"{item_label}.instruction", allow_empty=True),
174 }
175 if "when" in item:
176 layer["when"] = _require_string(item["when"], f"{item_label}.when")
177 result.append(layer)
178 return result
181def _keyed_by_code(tables: list[dict[str, Any]]) -> bool:
182 """Menus key their items on `code`; review layers on `id`. Same order as `structural_merge`."""
183 return all("code" in table for table in tables)
186def _require_keyed_tables(value: Any, label: str, key: str) -> list[dict[str, str]]:
187 if not isinstance(value, list):
188 raise RenderError(f"{label} must be a list of tables")
189 result: list[dict[str, str]] = []
190 seen: set[str] = set()
191 for index, item in enumerate(value):
192 item_label = f"{label}[{index}]"
193 if not isinstance(item, dict):
194 raise RenderError(f"{item_label} must be a table")
195 identifier = _require_string(item.get(key), f"{item_label}.{key}")
196 if identifier in seen:
197 raise RenderError(f"duplicate {label} {key} `{identifier}`")
198 seen.add(identifier)
199 result.append(
200 {field: _require_string(text, f"{item_label}.{field}", allow_empty=True) for field, text in item.items()}
201 )
202 return result
205def _load_sources(skill_dir: Path) -> dict[str, str]:
206 sources: dict[str, str] = {}
207 for candidate in sorted(skill_dir.rglob("*.md")):
208 if candidate.name == "SKILL.md":
209 continue
210 name = candidate.relative_to(skill_dir).as_posix()
211 path = candidate.resolve(strict=True)
212 if not path.is_relative_to(skill_dir):
213 raise RenderError(f"render source escapes skill directory: {name}")
214 if not path.is_file():
215 raise RenderError(f"render source is missing or not a file: {path}")
216 try:
217 sources[name] = path.read_text(encoding="utf-8")
218 except (OSError, UnicodeError) as error:
219 raise RenderError(f"failed to read render source {path}: {error}") from error
220 if "workflow.md" not in sources:
221 raise RenderError(f"render entry is missing: {skill_dir / 'workflow.md'}")
222 return sources
225def _load_assets(skill_dir: Path) -> dict[str, bytes]:
226 """Non-Markdown files under assets/ are copied into the generation unchanged, so a
227 customization value's `{skill-root}` (bound to the generation) reaches them too."""
228 assets: dict[str, bytes] = {}
229 for candidate in sorted((skill_dir / "assets").rglob("*")):
230 if candidate.suffix == ".md" or not candidate.is_file():
231 continue
232 name = candidate.relative_to(skill_dir).as_posix()
233 path = candidate.resolve(strict=True)
234 if not path.is_relative_to(skill_dir):
235 raise RenderError(f"render asset escapes skill directory: {name}")
236 try:
237 assets[name] = path.read_bytes()
238 except OSError as error:
239 raise RenderError(f"failed to read render asset {path}: {error}") from error
240 return assets
243def _resolve_config_value(value: Any, label: str, project_root: Path) -> str:
244 text = _require_string(value, label)
245 if "{project-root}" not in text:
246 return text
247 resolved = text.replace("{project-root}", project_root.as_posix())
248 if not Path(resolved).is_absolute():
249 raise RenderError(f"{label} must resolve to an absolute path: {resolved}")
250 return resolved
253def _find_config_values(data: Any, key: str, prefix: str = "") -> list[tuple[str, Any]]:
254 matches: list[tuple[str, Any]] = []
255 if not isinstance(data, dict):
256 return matches
257 for name, value in data.items():
258 path = f"{prefix}.{name}" if prefix else name
259 if name == key and not isinstance(value, (dict, list)):
260 matches.append((path, value))
261 matches.extend(_find_config_values(value, key, path))
262 return matches
265def _resolve_short_config(central: dict[str, Any], key: str, project_root: Path) -> tuple[str, str]:
266 matches = _find_config_values(central, key)
267 if not matches:
268 raise RenderError(f"missing config value `{key}`")
269 if len(matches) > 1:
270 paths = ", ".join(path for path, _ in matches)
271 raise RenderError(f"ambiguous config value `{key}` found at: {paths}")
272 path, value = matches[0]
273 return path, _resolve_config_value(value, f"config.{path}", project_root)
276def _format_markdown_list(items: list[str]) -> str:
277 if not items:
278 return "_None._"
279 rendered = []
280 for item in items:
281 lines = item.splitlines() or [""]
282 rendered.append("- " + lines[0])
283 rendered.extend(" " + line for line in lines[1:])
284 return "\n".join(rendered)
287def _format_review_layers(layers: list[dict[str, str]]) -> str:
288 active = [layer for layer in layers if layer["instruction"].strip()]
289 if not active:
290 return "No active review layers. HALT with blocking condition `no active review layers`."
291 sections = []
292 for layer in active:
293 section = [f"#### {layer['name']} (`{layer['id']}`)"]
294 if layer.get("when"):
295 section.extend(["", f"Run only when: {layer['when']}"])
296 section.extend(["", layer["instruction"].strip()])
297 sections.append("\n".join(section))
298 return "\n\n".join(sections)
301def _resolve_customization_value(value: Any, default: Any, label: str) -> Any:
302 """Validate an effective customization leaf against the shape of its shipped default."""
303 if isinstance(default, str):
304 allow_empty = not default.strip()
305 return _require_string(value, label, allow_empty=allow_empty)
306 if isinstance(default, list):
307 if default and all(isinstance(item, dict) for item in default):
308 if _keyed_by_code(default):
309 return _require_keyed_tables(value, label, "code")
310 return _require_review_layers(value, label)
311 return _require_string_list(value, label)
312 if isinstance(default, (bool, int, float, date, time)):
313 if type(value) is not type(default):
314 raise RenderError(f"{label} must be {type(default).__name__}, got {type(value).__name__}")
315 return value
316 raise RenderError(f"{label} has unsupported default type {type(default).__name__}")
319class _Text(str):
320 """A customization string. Looping over one is a template mistake, not a walk over its characters."""
322 def __new__(cls, value: str, label: str) -> _Text:
323 text = super().__new__(cls, value)
324 text.label = label
325 return text
327 def __iter__(self):
328 raise RenderError(f"`{self.label}` is a string, not a list")
331class _Derived:
332 """A name the renderer computes from the central config, when a template reaches it."""
334 def __init__(self, compute: Callable[[], Any]) -> None:
335 self.compute = compute
338class _MarkdownList(list):
339 """A string-list customization; inserted directly it renders as the Markdown list it always did."""
341 def __str__(self) -> str:
342 return _format_markdown_list(list(self))
345class _LayerList(list):
346 """A review-layer customization; inserted directly it renders as lens sections."""
348 def __str__(self) -> str:
349 return _format_review_layers(list(self))
352class _TableList(list):
353 """A code-keyed table customization such as a menu; inserted directly it renders one bullet per table."""
355 def __str__(self) -> str:
356 return _format_markdown_list(["\n".join(f"{key}: {text}" for key, text in table.items()) for table in self])
359def _bind_customization(value: Any, label: str, destination: Path) -> Any:
360 """Bind `{skill-root}` in customization prose to the generation and wrap lists for insertion."""
361 root = destination.as_posix()
362 if isinstance(value, str):
363 return _Text(value.replace("{skill-root}", root), label)
364 if isinstance(value, list):
365 if value and all(isinstance(item, dict) for item in value):
366 tables = [{key: text.replace("{skill-root}", root) for key, text in table.items()} for table in value]
367 return _TableList(tables) if _keyed_by_code(tables) else _LayerList(tables)
368 return _MarkdownList([item.replace("{skill-root}", root) for item in value])
369 return value
372class _Table:
373 """A dotted namespace over a TOML table. Names never hit Python attributes, so `workflow.items` is a lookup."""
375 def __init__(self, path: str) -> None:
376 self._path = path
378 def _child(self, name: str) -> str:
379 return f"{self._path}.{name}"
381 def _resolve(self, name: str) -> Any:
382 raise NotImplementedError
384 def __getattr__(self, name: str) -> Any:
385 if name.startswith("_"):
386 raise AttributeError(name)
387 return self._resolve(name)
389 def __getitem__(self, name: Any) -> Any:
390 if not isinstance(name, str):
391 raise RenderError(f"`{self._path}` is indexed by name, not {name!r}")
392 return self._resolve(name)
394 def __str__(self) -> str:
395 raise RenderError(f"`{self._path}` is a table, not a value")
398class _ConfigTable(_Table):
399 """`config.key` is the short lookup of one scalar anywhere in the central config; `config.a.b.c` is a path."""
401 def __init__(self, central: dict[str, Any], table: dict[str, Any], path: str, ctx: _RenderContext) -> None:
402 super().__init__(path)
403 self._central = central
404 self._table = table
405 self._ctx = ctx
407 def _resolve(self, name: str) -> Any:
408 if self._path == "config" and name not in self._table:
409 path, resolved = _resolve_short_config(self._central, name, self._ctx.project_root)
410 self._ctx.inputs[f"config.{path}"] = resolved
411 return _Text(resolved, f"config.{path}")
412 label = self._child(name)
413 if name not in self._table:
414 raise RenderError(f"missing config value `{label.removeprefix('config.')}`")
415 value = self._table[name]
416 if isinstance(value, dict):
417 return _ConfigTable(self._central, value, label, self._ctx)
418 resolved = _resolve_config_value(value, label, self._ctx.project_root)
419 self._ctx.inputs[label] = resolved
420 return _Text(resolved, label)
423class _CustomizationTable(_Table):
424 """The effective customization, each leaf validated against its `customize.toml` default."""
426 def __init__(self, defaults: dict[str, Any] | None, values: dict[str, Any], path: str, ctx: _RenderContext) -> None:
427 super().__init__(path)
428 self._defaults = defaults
429 self._values = values
430 self._ctx = ctx
432 def _resolve(self, name: str) -> Any:
433 path = self._child(name)
434 if self._defaults is None:
435 raise RenderError(f"`{path}` requires customize.toml")
436 if name not in self._defaults:
437 raise RenderError(f"missing customization parameter `{path}`")
438 if name not in self._values:
439 raise RenderError(f"missing customization value `{path}`")
440 default, value = self._defaults[name], self._values[name]
441 label = f"customization.{path}"
442 if isinstance(default, dict):
443 if not isinstance(value, dict):
444 raise RenderError(f"{label} must be a table, got {type(value).__name__}")
445 return _CustomizationTable(default, value, path, self._ctx)
446 resolved = _resolve_customization_value(value, default, label)
447 self._ctx.inputs[label] = resolved
448 return _bind_customization(resolved, label, self._ctx.destination)
451class _RenderContext:
452 """One rendering pass: the values it serves and the rendered() links each source makes."""
454 def __init__(
455 self,
456 *,
457 central: dict[str, Any],
458 defaults: dict[str, Any] | None,
459 customization: dict[str, Any],
460 source_names: set[str],
461 project_root: Path,
462 destination: Path,
463 ) -> None:
464 self.project_root = project_root
465 self.destination = destination
466 self.inputs: dict[str, Any] = {}
467 self.links: dict[str, set[str]] = {}
468 self._source_names = source_names
469 self.variables = {
470 "config": _ConfigTable(central, central, "config", self),
471 "workflow": _CustomizationTable(
472 None if defaults is None else defaults.get("workflow", {}),
473 customization.get("workflow", {}),
474 "workflow",
475 self,
476 ),
477 "agent": _CustomizationTable(
478 None if defaults is None else defaults.get("agent", {}),
479 customization.get("agent", {}),
480 "agent",
481 self,
482 ),
483 "rendered": self._rendered,
484 "halt": self._halt,
485 "initiative_folder": _Derived(self._initiative_folder),
486 }
487 self._central = central
489 def _initiative_folder(self) -> _Text:
490 """`core.output_folder`, extended by `/<core.active_initiative>` when an initiative is set."""
491 folder = str(self.variables["config"].core.output_folder)
492 core = self._central.get("core")
493 initiative = core.get("active_initiative") if isinstance(core, dict) else None
494 if initiative is not None:
495 folder = f"{folder}/{_require_string(initiative, 'config.core.active_initiative')}"
496 self.inputs["initiative_folder"] = folder
497 return _Text(folder, "initiative_folder")
499 @staticmethod
500 def _halt(message: Any) -> str:
501 """Let a template reject its inputs; the caller prefixes the source and line."""
502 raise RenderError(str(message))
504 @jinja2.pass_context
505 def _rendered(self, context: jinja2.runtime.Context, target: Any) -> str:
506 if not isinstance(target, str) or target not in self._source_names:
507 raise RenderError(f"rendered() targets undeclared source: {target}")
508 self.links.setdefault(context.name or "", set()).add(target)
509 return (self.destination / target).as_posix()
512class _Context(jinja2.runtime.Context):
513 """Reaching a derived name computes it, so only the values a template reads key its generation."""
515 def resolve_or_missing(self, key: str) -> Any:
516 value = super().resolve_or_missing(key)
517 return value.compute() if isinstance(value, _Derived) else value
520class _SourceLoader(jinja2.BaseLoader):
521 """Serve sources by name, and name them so template frames carry `source:line`."""
523 def __init__(self, sources: dict[str, str]) -> None:
524 self._sources = sources
526 def get_source(self, environment: jinja2.Environment, template: str) -> tuple[str, str, Any]:
527 if template not in self._sources:
528 raise jinja2.TemplateNotFound(template)
529 return self._sources[template], template, lambda: True
532def _template_location(error: BaseException, source_names: set[str]) -> str | None:
533 if isinstance(error, jinja2.TemplateSyntaxError):
534 return f"{error.name}:{error.lineno}" if error.name else None
535 location = None
536 traceback = error.__traceback__
537 while traceback is not None:
538 filename = traceback.tb_frame.f_code.co_filename
539 if filename in source_names:
540 location = f"{filename}:{traceback.tb_lineno}"
541 traceback = traceback.tb_next
542 return location
545def _render_sources(sources: dict[str, str], skill_dir: Path, context: _RenderContext) -> dict[str, str]:
546 """Render every source as a Jinja2 template against the context; return the non-empty outputs."""
547 # Skill sources name their bundled non-Markdown files (scripts, assets)
548 # through {skill-root}; those stay in the installed skill directory.
549 bound = {name: content.replace("{skill-root}", skill_dir.as_posix()) for name, content in sources.items()}
550 environment = jinja2.Environment(
551 loader=_SourceLoader(bound),
552 undefined=jinja2.StrictUndefined,
553 autoescape=False,
554 keep_trailing_newline=True,
555 trim_blocks=True,
556 lstrip_blocks=True,
557 )
558 environment.context_class = _Context
559 rendered: dict[str, str] = {}
560 for name in sources:
561 try:
562 rendered[name] = environment.get_template(name).render(context.variables)
563 except Exception as error:
564 location = _template_location(error, set(sources))
565 message = str(error) if isinstance(error, (RenderError, ConfigError, jinja2.TemplateError)) else repr(error)
566 raise RenderError(f"{location or name}: {message}") from error
567 # A source whose body renders to nothing is left out; links into it from survivors are broken.
568 omitted = {name for name, text in rendered.items() if not text.strip()}
569 if "workflow.md" in omitted:
570 raise RenderError("workflow.md: rendered empty")
571 for name in sorted(set(rendered) - omitted):
572 for target in sorted(context.links.get(name, set()) & omitted):
573 raise RenderError(f"{name}: rendered() targets omitted source: {target}")
574 return {name: text for name, text in rendered.items() if name not in omitted}
577def _verify_existing(destination: Path, manifest: dict[str, Any]) -> None:
578 manifest_path = destination / "manifest.json"
579 try:
580 existing = json.loads(manifest_path.read_text(encoding="utf-8"))
581 except (OSError, UnicodeError, json.JSONDecodeError) as error:
582 raise RenderError(f"corrupt existing generation {destination}: {error}") from error
583 if existing != manifest:
584 raise RenderError(f"generation collision or corruption at {destination}")
585 # Only the manifest's own files are verified. Anything else in the folder (Thumbs.db,
586 # editor swap files, sync conflict copies) is read by nobody and is left alone.
587 missing = sorted(name for name in manifest["outputs"] if not (destination / name).is_file())
588 if missing:
589 raise RenderError(
590 f"generation is missing rendered files: {', '.join(missing)} in {destination}; "
591 "deleting that folder is safe because the next run renders it again"
592 )
593 for name, expected_hash in manifest["outputs"].items():
594 try:
595 actual_hash = _hash_bytes((destination / name).read_bytes())
596 except OSError as error:
597 raise RenderError(f"failed to verify {destination / name}: {error}") from error
598 if actual_hash != expected_hash:
599 raise RenderError(f"generation output hash mismatch: {destination / name}")
602def _publish(destination: Path, outputs: dict[str, bytes], manifest: dict[str, Any]) -> None:
603 destination.parent.mkdir(parents=True, exist_ok=True)
604 if destination.exists():
605 _verify_existing(destination, manifest)
606 return
607 # One attempt, not mkdtemp: on Windows, older Pythons' mkdtemp takes "access denied"
608 # for a name collision and tries the next name, some two billion times. The render
609 # would hang in a folder it cannot write to instead of halting.
610 staging = destination.parent / f".staging-{os.urandom(8).hex()}"
611 staging.mkdir(mode=0o700)
612 try:
613 for name, content in outputs.items():
614 path = staging / name
615 path.parent.mkdir(parents=True, exist_ok=True)
616 path.write_bytes(content)
617 (staging / "manifest.json").write_bytes(
618 json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True, default=_json_scalar).encode("utf-8")
619 + b"\n"
620 )
621 try:
622 os.rename(staging, destination)
623 except OSError:
624 if destination.exists():
625 _verify_existing(destination, manifest)
626 else:
627 raise
628 finally:
629 if staging.exists():
630 shutil.rmtree(staging, ignore_errors=True)
633def render(
634 project_root: Path, skill_dir: Path, *, overrides: Path | None = None, assignments: list[str] | None = None
635) -> Path:
636 project_root = project_root.resolve(strict=True)
637 skill_dir = skill_dir.resolve(strict=True)
638 if not (project_root / "_bmad").is_dir():
639 raise RenderError(f"project root does not contain _bmad/: {project_root}")
641 sources = _load_sources(skill_dir)
642 assets = _load_assets(skill_dir)
643 central = load_central_config(project_root)
644 has_customization = bool(overrides is not None or assignments) or (skill_dir / "customize.toml").is_file()
645 defaults = load_toml(skill_dir / "customize.toml", required=True) if has_customization else None
646 _check_persistent_layers(project_root, skill_dir, defaults)
647 customization = load_customization(project_root, skill_dir) if has_customization else {}
648 supplied: set[str] = set()
649 if defaults is not None:
650 file_layer, command_layer = _invocation_customization(defaults, overrides, assignments or [])
651 customization = structural_merge(structural_merge(customization, file_layer), command_layer)
652 supplied = _leaf_paths(file_layer) | _leaf_paths(command_layer)
654 # Copied assets key the generation like rendered sources do, so an edited asset is not served stale.
655 source_hashes = {name: _hash_bytes(content.encode("utf-8")) for name, content in sources.items()}
656 source_hashes |= {name: _hash_bytes(content) for name, content in assets.items()}
657 root_hash = _hash_bytes(str(project_root).encode("utf-8"))[:12]
658 slug = re.sub(r"[^a-z0-9]+", "-", project_root.name.lower()).strip("-") or "project"
659 slug = slug[:80].rstrip("-") or "project"
660 namespace = project_root / "_bmad" / "render" / skill_dir.name / f"{slug}-{root_hash}"
662 def render_pass(destination: Path) -> tuple[_RenderContext, dict[str, str]]:
663 context = _RenderContext(
664 central=central,
665 defaults=defaults,
666 customization=customization,
667 source_names=set(sources),
668 project_root=project_root,
669 destination=destination,
670 )
671 return context, _render_sources(sources, skill_dir, context)
673 # The generation path is keyed by the values the templates reach, and the
674 # templates insert that path, so a first pass against a placeholder
675 # destination collects the inputs and the real pass renders the output.
676 probe, _ = render_pass(namespace / "pending")
677 # An override may only name a key some template actually read; values are validated where consumed.
678 unused = sorted(path for path in supplied if f"customization.{path}" not in probe.inputs)
679 if unused:
680 raise RenderError(f"invocation override not used by this render: {', '.join(unused)}")
681 # Store TOML date/time inputs in the same JSON representation used on disk.
682 input_values = json.loads(_canonical_json(probe.inputs))
683 renderer_hash = _hash_bytes(Path(__file__).read_bytes())
684 identity = {
685 "project_root": str(project_root),
686 "skill_root": str(skill_dir),
687 "renderer_sha256": renderer_hash,
688 "jinja2_version": jinja2.__version__,
689 "resolved_values": input_values,
690 "source_sha256": source_hashes,
691 }
692 generation_hash = _hash_bytes(_canonical_json(identity))[:20]
693 destination = namespace / generation_hash
694 _, rendered = render_pass(destination)
695 outputs = {name: content.encode("utf-8") for name, content in rendered.items()} | assets
696 output_hashes = {name: _hash_bytes(content) for name, content in outputs.items()}
697 manifest = {
698 "schema_version": 1,
699 "skill": skill_dir.name,
700 "project_root": str(project_root),
701 "project_slug": slug,
702 "root_hash": root_hash,
703 "generation_hash": generation_hash,
704 "inputs": identity,
705 "outputs": output_hashes,
706 }
707 _publish(destination, outputs, manifest)
708 return destination / "workflow.md"
711def report_owed_setup(skill_dir: Path, project_root: Path, roots: tuple[Path, ...] = ()) -> None:
712 # Runs before rendering so the note lands ahead of the printed workflow,
713 # and still shows when rendering halts. It must never fail the render.
714 try:
715 import setup_check
716 except Exception:
717 return
718 setup_check.report(skill_dir, project_root, roots)
721def main() -> int:
722 parser = _ArgumentParser(description=__doc__)
723 parser.add_argument("--project-root", required=True)
724 parser.add_argument("--skill", required=True)
725 parser.add_argument("--overrides", type=Path, help="invocation-only customization TOML file")
726 parser.add_argument("--set", dest="assignments", action="append", default=[], metavar="KEY=VALUE")
727 parser.add_argument(
728 "--root",
729 type=Path,
730 action="append",
731 default=[],
732 help="a skills root the host has active, in precedence order, project roots first (repeatable)",
733 )
734 reconfigure = getattr(sys.stdout, "reconfigure", None)
735 if reconfigure is not None:
736 reconfigure(encoding="utf-8")
737 try:
738 args = parser.parse_args()
739 report_owed_setup(Path(args.skill).resolve(), Path(args.project_root).resolve(), tuple(args.root))
740 entry = render(
741 Path(args.project_root), Path(args.skill), overrides=args.overrides, assignments=args.assignments
742 )
743 # Printed, not named: a file the model is told to read may be read only in part.
744 workflow = entry.read_text(encoding="utf-8")
745 except (ConfigError, RenderError, OSError, UnicodeError, ValueError) as error:
746 sys.stdout.write(f"HALT: {' '.join(str(error).splitlines())}\n")
747 return 1
748 sys.stdout.write(workflow)
749 return 0
752if __name__ == "__main__":
753 if sys.platform == "win32":
754 # Piped output on Windows defaults to a legacy code page, not UTF-8.
755 sys.stdout.reconfigure(encoding="utf-8")
756 sys.stderr.reconfigure(encoding="utf-8")
757 raise SystemExit(main())