From 5df938805e31454910263d54e34110c04c94cdd9 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Fri, 3 Jul 2026 10:46:19 -0400 Subject: [PATCH] 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. --- scripts/aggregate_directives.py | 58 +++++++++++++++++++++++++-------- 1 file changed, 45 insertions(+), 13 deletions(-) diff --git a/scripts/aggregate_directives.py b/scripts/aggregate_directives.py index 1a32eef5..9e88e03f 100644 --- a/scripts/aggregate_directives.py +++ b/scripts/aggregate_directives.py @@ -23,7 +23,9 @@ ENTRY_RE = re.compile(r"^\s*-\s+(?P[A-Za-z0-9_]+)\s*:\s*(?P\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)):