From 82468611ba4ae684e335ecd049f3b142409dbe8b Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 14:14:22 -0400 Subject: [PATCH] feat(aggregate): add Inherits resolution + Non-directive context to preset parser --- scripts/aggregate_directives.py | 193 ++++++++++++++----- tests/test_aggregate_directives_presets.py | 208 +++++++++++++++++++++ 2 files changed, 359 insertions(+), 42 deletions(-) create mode 100644 tests/test_aggregate_directives_presets.py diff --git a/scripts/aggregate_directives.py b/scripts/aggregate_directives.py index 9e88e03f..88134b87 100644 --- a/scripts/aggregate_directives.py +++ b/scripts/aggregate_directives.py @@ -1,10 +1,19 @@ #!/usr/bin/env python3 -"""Aggregate directive bodies from a preset markdown file.""" +"""Aggregate directive bodies from a preset markdown file. + +Supports three preset sections: + ## Inherits — name of parent preset (resolved recursively, 3-level max) + ## Directives to warm — list of : + ## Non-directive context — list of : + +Child directives + non-directive files override parent on name/path conflict. +""" from __future__ import annotations import argparse import re import sys +from dataclasses import dataclass from pathlib import Path if hasattr(sys.stdout, "reconfigure"): @@ -19,36 +28,113 @@ if hasattr(sys.stderr, "reconfigure"): pass REPO_ROOT = Path(__file__).resolve().parent.parent -ENTRY_RE = re.compile(r"^\s*-\s+(?P[A-Za-z0-9_]+)\s*:\s*(?P\S+)\s*$") +ENTRY_RE = re.compile(r"^\s*-\s+(?P[^:]+):\s+(?P\S+)\s*$") EXPECTED_PATTERN_TEMPLATE = r"^conductor/directives/{name}/v\d+\.md$" +MAX_INHERIT_DEPTH = 3 -def parse_preset(preset_path: Path, root: Path | None = None) -> list[tuple[str, Path]]: +@dataclass(frozen=True, slots=True) +class PresetSections: + inherits: str + directives: list[tuple[str, Path]] + non_directive_context: list[tuple[str, Path]] + + +def parse_preset_sections(preset_path: Path) -> PresetSections: + """Parse a preset markdown file into Inherits, Directives, Non-directive context.""" + if not preset_path.exists(): + raise FileNotFoundError("preset file not found: " + str(preset_path)) + text = preset_path.read_text(encoding="utf-8") + inherits: str = "" + directives: list[tuple[str, Path]] = [] + non_directive: list[tuple[str, Path]] = [] + current_section: str = "" + for line in text.splitlines(): + if line.startswith("## Inherits"): + current_section = "inherits" + continue + if line.startswith("## Directives to warm"): + current_section = "directives" + continue + if line.startswith("## Non-directive context"): + current_section = "non_directive" + continue + if line.startswith("## ") or line.startswith("# "): + current_section = "" + continue + if current_section == "inherits" and line.strip(): + inherits = line.strip() + elif current_section in ("directives", "non_directive"): + m = ENTRY_RE.match(line) + if m: + name = m.group("name") + rel_path = Path(m.group("path")) + if current_section == "directives": + directives.append((name, rel_path)) + else: + non_directive.append((name, rel_path)) + return PresetSections(inherits=inherits, directives=directives, non_directive_context=non_directive) + + +def resolve_inheritance( + preset_path: Path, + root: Path | None = None, + _visited: frozenset[str] | None = None, + _depth: int = 0, +) -> PresetSections: + """Resolve a preset, recursively flattening parent directives + non-directive files.""" 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) - if not m: - continue - name = m.group("name") - rel_path = m.group("path") - abs_path = (root / rel_path).resolve() - entries.append((name, abs_path)) - return entries + p = preset_path if preset_path.is_absolute() else (root / preset_path).resolve() + if _visited is None: + _visited = frozenset() + key = str(p) + if key in _visited: + raise ValueError("inheritance loop detected at " + key) + if _depth > MAX_INHERIT_DEPTH: + raise ValueError("inheritance depth exceeds " + str(MAX_INHERIT_DEPTH) + " at " + key) + sections = parse_preset_sections(p) + if not sections.inherits: + return sections + parent_name = sections.inherits + parent_path = (root / "conductor" / "directives" / "presets" / (parent_name + ".md")).resolve() + if not parent_path.exists(): + raise FileNotFoundError("parent preset not found: " + str(parent_path)) + parent = resolve_inheritance(parent_path, root=root, _visited=_visited | {key}, _depth=_depth + 1) + merged_directives: list[tuple[str, Path]] = [] + child_names: set[str] = {n for n, _ in sections.directives} + for name, path in parent.directives: + if name not in child_names: + merged_directives.append((name, path)) + for name, path in sections.directives: + merged_directives.append((name, path)) + merged_non_directive: list[tuple[str, Path]] = [] + child_paths: set[Path] = {p for _, p in sections.non_directive_context} + for desc, path in parent.non_directive_context: + if path not in child_paths: + merged_non_directive.append((desc, path)) + for desc, path in sections.non_directive_context: + merged_non_directive.append((desc, path)) + return PresetSections(inherits="", directives=merged_directives, non_directive_context=merged_non_directive) -def aggregate_directives(preset_path: str | Path, *, max_chars: int = 0, project_root: str | Path | None = None) -> str: +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. + Read preset at preset_path, resolve inheritance, read each directive's v1.md + and each non-directive-context file, concatenate into a single text blob. - 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 format: + ## Inherits + ## Directives to warm (list of : ) + ## Non-directive context (list of : ) - 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) + Directives appear first (same ===== banner format as before), then a + NON-DIRECTIVE CONTEXT section appends each non-directive file's full body. """ root = Path(project_root).resolve() if project_root is not None else REPO_ROOT p = Path(preset_path) @@ -56,43 +142,66 @@ def aggregate_directives(preset_path: str | Path, *, max_chars: int = 0, project 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: - raise ValueError("no directive entries found in " + str(p)) + sections = resolve_inheritance(p, root=root) + if not sections.directives and not sections.non_directive_context: + raise ValueError("no directive or non-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)) + directive_bodies: list[tuple[str, str]] = [] + for name, rel in sections.directives: + abs_path = (root / rel).resolve() + if not abs_path.exists(): + missing.append((name, abs_path)) continue try: - rel = path.relative_to(root).as_posix() + rel_str = abs_path.relative_to(root).as_posix() except ValueError: - rel = str(path) + rel_str = str(abs_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) + if not re.match(pattern, rel_str): + print("WARNING: " + name + ": path " + str(abs_path) + " does not match pattern conductor/directives/" + name + "/vN.md; including anyway", file=sys.stderr) try: - body = path.read_text(encoding="utf-8") + body = abs_path.read_text(encoding="utf-8") except UnicodeDecodeError as e: - raise ValueError(name + ": UTF-8 decode error in " + str(path) + ": " + str(e)) from e - bodies.append((name, body)) + raise ValueError(name + ": UTF-8 decode error in " + str(abs_path) + ": " + str(e)) from e + directive_bodies.append((name, body)) if missing: names = ", ".join(n for n, _ in missing) raise FileNotFoundError("missing v1.md files for directives: " + names) + non_directive_bodies: list[tuple[str, str]] = [] + for desc, rel in sections.non_directive_context: + abs_path = (root / rel).resolve() + if not abs_path.exists(): + print("WARNING: non-directive context file not found: " + str(abs_path), file=sys.stderr) + continue + try: + body = abs_path.read_text(encoding="utf-8") + except UnicodeDecodeError as e: + raise ValueError(desc + ": UTF-8 decode error in " + str(abs_path) + ": " + str(e)) from e + non_directive_bodies.append((desc, body)) out_lines: list[str] = [] - for name, body in bodies: + for name, body in directive_bodies: out_lines.append("=" * 40) out_lines.append(name) out_lines.append("=" * 40) out_lines.append("") - out_lines.append(body.rstrip(chr(10))) + out_lines.append(body.rstrip("\n")) out_lines.append("") out_lines.append("") - rendered = chr(10).join(out_lines) + if non_directive_bodies: + out_lines.append("=" * 40) + out_lines.append("NON-DIRECTIVE CONTEXT") + out_lines.append("=" * 40) + out_lines.append("") + for desc, body in non_directive_bodies: + out_lines.append("--- " + desc + " ---") + out_lines.append("") + out_lines.append(body.rstrip("\n")) + out_lines.append("") + out_lines.append("") + rendered = "\n".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]" + suffix = "\n\n[truncated: showing " + str(max_chars) + " of " + str(len(rendered)) + " chars; omit max_chars for full text]" rendered = truncated + suffix return rendered @@ -105,11 +214,11 @@ def aggregate(preset_path: Path, output_path: Path | None) -> int: return 1 if output_path: output_path.write_text(rendered, encoding="utf-8") - print("Wrote " + str(rendered.count("=" * 40) // 2) + " directives to " + str(output_path), file=sys.stderr) + print("Wrote " + str(rendered.count("=" * 40) // 2) + " sections to " + str(output_path), file=sys.stderr) else: sys.stdout.write(rendered) - if not rendered.endswith(chr(10)): - sys.stdout.write(chr(10)) + if not rendered.endswith("\n"): + sys.stdout.write("\n") return 0 diff --git a/tests/test_aggregate_directives_presets.py b/tests/test_aggregate_directives_presets.py new file mode 100644 index 00000000..d2a1ea6c --- /dev/null +++ b/tests/test_aggregate_directives_presets.py @@ -0,0 +1,208 @@ +"""Tests for aggregate_directives preset inheritance and non-directive context.""" +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +# Add scripts/ to sys.path so we can import aggregate_directives +SCRIPTS_DIR = Path(__file__).resolve().parent.parent / "scripts" +if str(SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPTS_DIR)) + +from aggregate_directives import parse_preset_sections, resolve_inheritance, aggregate_directives + + +def _preset_dir(root: Path) -> Path: + """Return the preset directory under root, creating it if needed.""" + d = root / "conductor" / "directives" / "presets" + d.mkdir(parents=True, exist_ok=True) + return d + + +class TestParsePresetSections: + """Test the new section-based parser that extracts Inherits, Directives, Non-directive context.""" + + def test_parse_returns_three_sections(self, tmp_path: Path): + preset = tmp_path / "test_preset.md" + preset.write_text("""# Preset: test + +## Inherits +baseline + +## Directives to warm +Read each file below before any action. + +- foo: conductor/directives/foo/v1.md +- bar: conductor/directives/bar/v1.md + +## Non-directive context +Read each file below as full documents. + +- project rules: AGENTS.md + +## Notes +test +""", encoding="utf-8") + sections = parse_preset_sections(preset) + assert sections.inherits == "baseline" + assert len(sections.directives) == 2 + assert sections.directives[0] == ("foo", Path("conductor/directives/foo/v1.md")) + assert sections.directives[1] == ("bar", Path("conductor/directives/bar/v1.md")) + assert len(sections.non_directive_context) == 1 + assert sections.non_directive_context[0] == ("project rules", Path("AGENTS.md")) + + def test_parse_preset_without_inherits(self, tmp_path: Path): + preset = tmp_path / "test_preset.md" + preset.write_text("""# Preset: test + +## Directives to warm +- foo: conductor/directives/foo/v1.md + +## Non-directive context +- doc: AGENTS.md +""", encoding="utf-8") + sections = parse_preset_sections(preset) + assert sections.inherits == "" + assert len(sections.directives) == 1 + assert len(sections.non_directive_context) == 1 + + def test_parse_preset_without_non_directive_context(self, tmp_path: Path): + preset = tmp_path / "test_preset.md" + preset.write_text("""# Preset: test + +## Directives to warm +- foo: conductor/directives/foo/v1.md +""", encoding="utf-8") + sections = parse_preset_sections(preset) + assert sections.inherits == "" + assert len(sections.directives) == 1 + assert sections.non_directive_context == [] + + +class TestResolveInheritance: + """Test that resolve_inheritance merges parent directives into child, dedup on name.""" + + def test_child_directives_plus_parent_directives(self, tmp_path: Path): + pdir = _preset_dir(tmp_path) + (pdir / "parent.md").write_text("""# Preset: parent + +## Directives to warm +- alpha: conductor/directives/alpha/v1.md +- beta: conductor/directives/beta/v1.md +""", encoding="utf-8") + (pdir / "child.md").write_text("""# Preset: child + +## Inherits +parent + +## Directives to warm +- beta: conductor/directives/beta/v1.md +- gamma: conductor/directives/gamma/v1.md +""", encoding="utf-8") + merged = resolve_inheritance(pdir / "child.md", root=tmp_path) + names = [n for n, _ in merged.directives] + assert set(names) == {"alpha", "beta", "gamma"} + assert names.count("beta") == 1 + assert len(merged.non_directive_context) == 0 + + def test_child_wins_on_directive_conflict(self, tmp_path: Path): + pdir = _preset_dir(tmp_path) + (pdir / "parent.md").write_text("""# Preset: parent + +## Directives to warm +- shared: conductor/directives/shared_parent/v1.md +""", encoding="utf-8") + (pdir / "child.md").write_text("""# Preset: child + +## Inherits +parent + +## Directives to warm +- shared: conductor/directives/shared_child/v1.md +""", encoding="utf-8") + merged = resolve_inheritance(pdir / "child.md", root=tmp_path) + paths = [p for _, p in merged.directives] + assert Path("conductor/directives/shared_child/v1.md") in paths + assert Path("conductor/directives/shared_parent/v1.md") not in paths + + def test_non_directive_context_merges_dedup_on_path(self, tmp_path: Path): + pdir = _preset_dir(tmp_path) + (pdir / "parent.md").write_text("""# Preset: parent + +## Directives to warm +- alpha: conductor/directives/alpha/v1.md + +## Non-directive context +- rules: AGENTS.md +- product: conductor/product.md +""", encoding="utf-8") + (pdir / "child.md").write_text("""# Preset: child + +## Inherits +parent + +## Directives to warm +- beta: conductor/directives/beta/v1.md + +## Non-directive context +- rules: AGENTS.md +- tech: conductor/tech-stack.md +""", encoding="utf-8") + merged = resolve_inheritance(pdir / "child.md", root=tmp_path) + paths = [p for _, p in merged.non_directive_context] + assert Path("AGENTS.md") in paths + assert Path("conductor/product.md") in paths + assert Path("conductor/tech-stack.md") in paths + assert paths.count(Path("AGENTS.md")) == 1 + + def test_three_level_inheritance(self, tmp_path: Path): + pdir = _preset_dir(tmp_path) + (pdir / "grandparent.md").write_text("""# Preset: grandparent + +## Directives to warm +- alpha: conductor/directives/alpha/v1.md +""", encoding="utf-8") + (pdir / "parent.md").write_text("""# Preset: parent + +## Inherits +grandparent + +## Directives to warm +- beta: conductor/directives/beta/v1.md +""", encoding="utf-8") + (pdir / "child.md").write_text("""# Preset: child + +## Inherits +parent + +## Directives to warm +- gamma: conductor/directives/gamma/v1.md +""", encoding="utf-8") + merged = resolve_inheritance(pdir / "child.md", root=tmp_path) + names = [n for n, _ in merged.directives] + assert "alpha" in names + assert "beta" in names + assert "gamma" in names + + def test_inheritance_loop_detected(self, tmp_path: Path): + pdir = _preset_dir(tmp_path) + (pdir / "a.md").write_text("""# Preset: a + +## Inherits +b + +## Directives to warm +- alpha: conductor/directives/alpha/v1.md +""", encoding="utf-8") + (pdir / "b.md").write_text("""# Preset: b + +## Inherits +a + +## Directives to warm +- beta: conductor/directives/beta/v1.md +""", encoding="utf-8") + with pytest.raises(ValueError, match="inheritance loop"): + resolve_inheritance(pdir / "a.md", root=tmp_path) \ No newline at end of file