refactor(aggregate_directives): expose importable aggregate_directives(preset_path, max_chars, project_root) function

Wraps the body-building core in a public aggregate_directives() function that
returns the rendered string and raises FileNotFoundError/ValueError on errors.
The existing aggregate() CLI wrapper catches those and preserves the prior
stdout/file output behavior. parse_preset now accepts an optional root param so
directive paths inside the preset resolve against a caller-supplied
project_root instead of always REPO_ROOT. No behavior change for the CLI; the
new function is the API surface used by the upcoming MCP server tool.
This commit is contained in:
ed
2026-07-03 10:46:19 -04:00
parent 7d7f88f823
commit 5df938805e
+45 -13
View File
@@ -23,7 +23,9 @@ ENTRY_RE = re.compile(r"^\s*-\s+(?P<name>[A-Za-z0-9_]+)\s*:\s*(?P<path>\S+)\s*$"
EXPECTED_PATTERN_TEMPLATE = r"^conductor/directives/{name}/v\d+\.md$"
def parse_preset(preset_path: Path) -> list[tuple[str, Path]]:
def parse_preset(preset_path: Path, root: Path | None = None) -> list[tuple[str, Path]]:
if root is None:
root = REPO_ROOT
entries: list[tuple[str, Path]] = []
for line in preset_path.read_text(encoding="utf-8").splitlines():
m = ENTRY_RE.match(line)
@@ -31,36 +33,53 @@ def parse_preset(preset_path: Path) -> list[tuple[str, Path]]:
continue
name = m.group("name")
rel_path = m.group("path")
abs_path = (REPO_ROOT / rel_path).resolve()
abs_path = (root / rel_path).resolve()
entries.append((name, abs_path))
return entries
def aggregate(preset_path: Path, output_path: Path | None) -> int:
entries = parse_preset(preset_path)
def aggregate_directives(preset_path: str | Path, *, max_chars: int = 0, project_root: str | Path | None = None) -> str:
"""
Read preset at preset_path, resolve each directive's v1.md, concatenate bodies.
Reads ONLY v1.md (never meta.md). Returns the concatenated string.
Raises FileNotFoundError if the preset or any referenced v1.md is missing,
ValueError if the preset is empty or any v1.md fails UTF-8 decoding.
preset_path : path to the presets markdown file (absolute or relative to project_root)
max_chars : if > 0, truncate the rendered body to this many chars and append a suffix note
project_root : override for the project root (defaults to the module's REPO_ROOT)
"""
root = Path(project_root).resolve() if project_root is not None else REPO_ROOT
p = Path(preset_path)
if not p.is_absolute():
p = (root / p).resolve()
if not p.exists():
raise FileNotFoundError("preset file not found: " + str(p))
entries = parse_preset(p, root=root)
if not entries:
print("ERROR: no directive entries found in " + str(preset_path), file=sys.stderr)
return 1
raise ValueError("no directive entries found in " + str(p))
missing: list[tuple[str, Path]] = []
bodies: list[tuple[str, str]] = []
for name, path in entries:
if not path.exists():
missing.append((name, path))
continue
rel = path.relative_to(REPO_ROOT).as_posix()
try:
rel = path.relative_to(root).as_posix()
except ValueError:
rel = str(path)
pattern = EXPECTED_PATTERN_TEMPLATE.format(name=re.escape(name))
if not re.match(pattern, rel):
print("WARNING: " + name + ": path " + str(path) + " does not match pattern conductor/directives/" + name + "/vN.md; including anyway", file=sys.stderr)
try:
body = path.read_text(encoding="utf-8")
except UnicodeDecodeError as e:
print("ERROR: " + name + ": UTF-8 decode error in " + str(path) + ": " + str(e), file=sys.stderr)
continue
raise ValueError(name + ": UTF-8 decode error in " + str(path) + ": " + str(e)) from e
bodies.append((name, body))
if missing:
for name, path in missing:
print("ERROR: missing file for directive '" + name + "': " + str(path), file=sys.stderr)
return 1
names = ", ".join(n for n, _ in missing)
raise FileNotFoundError("missing v1.md files for directives: " + names)
out_lines: list[str] = []
for name, body in bodies:
out_lines.append("=" * 40)
@@ -71,9 +90,22 @@ def aggregate(preset_path: Path, output_path: Path | None) -> int:
out_lines.append("")
out_lines.append("")
rendered = chr(10).join(out_lines)
if max_chars > 0 and len(rendered) > max_chars:
truncated = rendered[:max_chars]
suffix = chr(10) + chr(10) + "[truncated: showing " + str(max_chars) + " of " + str(len(rendered)) + " chars; omit max_chars for full text]"
rendered = truncated + suffix
return rendered
def aggregate(preset_path: Path, output_path: Path | None) -> int:
try:
rendered = aggregate_directives(preset_path)
except (FileNotFoundError, ValueError) as e:
print("ERROR: " + str(e), file=sys.stderr)
return 1
if output_path:
output_path.write_text(rendered, encoding="utf-8")
print("Wrote " + str(len(bodies)) + " directives to " + str(output_path), file=sys.stderr)
print("Wrote " + str(rendered.count("=" * 40) // 2) + " directives to " + str(output_path), file=sys.stderr)
else:
sys.stdout.write(rendered)
if not rendered.endswith(chr(10)):