| 1 | #!/usr/bin/env python3 |
| 2 | """Render the topic-aware analysis prompt template with values from squadscope.topic.yml. |
| 3 | |
| 4 | Usage: |
| 5 | python scripts/render_topic_prompt.py [--config PATH] [--output PATH] |
| 6 | |
| 7 | If --config is not provided, looks for squadscope.topic.yml in the repo root. |
| 8 | If --output is not provided, prints to stdout. |
| 9 | """ |
| 10 | |
| 11 | from __future__ import annotations |
| 12 | |
| 13 | import argparse |
| 14 | import sys |
| 15 | from pathlib import Path |
| 16 | |
| 17 | # Use PyYAML if available, otherwise fall back to a minimal inline parser |
| 18 | try: |
| 19 | import yaml # type: ignore[import-untyped] |
| 20 | |
| 21 | def _load_yaml(path: Path) -> dict: |
| 22 | with open(path) as f: |
| 23 | return yaml.safe_load(f) or {} |
| 24 | |
| 25 | except ImportError: |
| 26 | # Minimal YAML subset parser for simple key: value files |
| 27 | def _load_yaml(path: Path) -> dict: # type: ignore[misc] |
| 28 | result: dict = {} |
| 29 | with open(path) as f: |
| 30 | for line in f: |
| 31 | line = line.strip() |
| 32 | if not line or line.startswith("#"): |
| 33 | continue |
| 34 | if ":" in line: |
| 35 | key, _, value = line.partition(":") |
| 36 | key = key.strip() |
| 37 | value = value.strip().strip('"').strip("'") |
| 38 | result[key] = value |
| 39 | return result |
| 40 | |
| 41 | |
| 42 | def find_repo_root() -> Path: |
| 43 | """Walk up from CWD to find the git repo root.""" |
| 44 | current = Path.cwd() |
| 45 | while current != current.parent: |
| 46 | if (current / ".git").exists(): |
| 47 | return current |
| 48 | current = current.parent |
| 49 | return Path.cwd() |
| 50 | |
| 51 | |
| 52 | def load_topic_config(config_path: Path | None) -> dict | None: |
| 53 | """Load topic configuration from YAML. Returns None if not found.""" |
| 54 | if config_path and config_path.exists(): |
| 55 | return _load_yaml(config_path) |
| 56 | |
| 57 | # Try default location |
| 58 | root = find_repo_root() |
| 59 | default_path = root / "squadscope.topic.yml" |
| 60 | if default_path.exists(): |
| 61 | return _load_yaml(default_path) |
| 62 | |
| 63 | return None |
| 64 | |
| 65 | |
| 66 | def load_wisdom(topic_id: str | None) -> str: |
| 67 | """Load per-topic wisdom file if it exists.""" |
| 68 | if not topic_id: |
| 69 | return "" |
| 70 | |
| 71 | # Reject topic IDs that could escape the topics/ directory. |
| 72 | import re |
| 73 | |
| 74 | if not re.fullmatch(r"[a-z0-9][a-z0-9\-_]{0,63}", topic_id): |
| 75 | return "" |
| 76 | |
| 77 | root = find_repo_root() |
| 78 | wisdom_path = root / "topics" / topic_id / "wisdom.md" |
| 79 | # Containment check: ensure the resolved path stays inside topics/ |
| 80 | topics_root = (root / "topics").resolve() |
| 81 | try: |
| 82 | wisdom_path.resolve().relative_to(topics_root) |
| 83 | except ValueError: |
| 84 | return "" |
| 85 | # Cap injected wisdom to 8 KiB to prevent prompt bloat from large/poisoned files. |
| 86 | _MAX_WISDOM_BYTES = 8192 |
| 87 | if wisdom_path.exists(): |
| 88 | content = wisdom_path.read_text(encoding="utf-8").strip() |
| 89 | if len(content.encode("utf-8")) > _MAX_WISDOM_BYTES: |
| 90 | content = content[:_MAX_WISDOM_BYTES] + "\n…[truncated]" |
| 91 | return content |
| 92 | |
| 93 | return "" |
| 94 | |
| 95 | |
| 96 | def _load_sanitize_text(): |
| 97 | try: |
| 98 | from scripts.sanitize_repo_content import sanitize_text as _sanitize_text |
| 99 | except (ImportError, ModuleNotFoundError): |
| 100 | scripts_dir = Path(__file__).resolve().parent |
| 101 | if str(scripts_dir) not in sys.path: |
| 102 | sys.path.insert(0, str(scripts_dir)) |
| 103 | from sanitize_repo_content import sanitize_text as _sanitize_text |
| 104 | return _sanitize_text |
| 105 | |
| 106 | |
| 107 | def render_template(template: str, topic_config: dict | None) -> str: |
| 108 | """Render the topic-aware prompt template with config values. |
| 109 | |
| 110 | Handles conditional blocks: |
| 111 | {{#IF_TOPIC}}...{{/IF_TOPIC}} — included only when topic config is present |
| 112 | {{#IF_NO_TOPIC}}...{{/IF_NO_TOPIC}} — included only when topic config is absent |
| 113 | """ |
| 114 | has_topic = topic_config is not None and bool( |
| 115 | topic_config.get("id") or topic_config.get("name") |
| 116 | ) |
| 117 | |
| 118 | if has_topic: |
| 119 | import re as _re |
| 120 | |
| 121 | topic_id = topic_config.get("id", "") |
| 122 | # Coerce topic_id to string to avoid TypeError in re.fullmatch |
| 123 | if not isinstance(topic_id, str): |
| 124 | topic_id = str(topic_id) if topic_id is not None else "" |
| 125 | topic_name = topic_config.get("name", "") |
| 126 | topic_description = topic_config.get("description", "") |
| 127 | |
| 128 | # Validate topic_id before prompt injection (same regex as load_wisdom) |
| 129 | if not _re.fullmatch(r"[a-z0-9][a-z0-9\-_]{0,63}", topic_id): |
| 130 | topic_id = "" |
| 131 | |
| 132 | wisdom_content = load_wisdom(topic_id) |
| 133 | |
| 134 | # Sanitize user-controlled topic fields |
| 135 | sanitize_text = _load_sanitize_text() |
| 136 | |
| 137 | topic_name = sanitize_text(topic_name, max_length=200, label="topic_name") |
| 138 | topic_description = sanitize_text( |
| 139 | topic_description, max_length=500, label="topic_description" |
| 140 | ) |
| 141 | |
| 142 | # Remove IF_NO_TOPIC blocks |
| 143 | rendered = _remove_blocks(template, "IF_NO_TOPIC") |
| 144 | # Keep IF_TOPIC block contents |
| 145 | rendered = _keep_blocks(rendered, "IF_TOPIC") |
| 146 | |
| 147 | # Replace placeholders |
| 148 | rendered = rendered.replace("{{TOPIC_ID}}", topic_id) |
| 149 | rendered = rendered.replace("{{TOPIC_NAME}}", topic_name) |
| 150 | rendered = rendered.replace("{{TOPIC_DESCRIPTION}}", topic_description) |
| 151 | # Sanitize boundary markers in wisdom content to prevent fence escape |
| 152 | try: |
| 153 | from scripts.sanitize_repo_content import _escape_untrusted_boundaries |
| 154 | except (ImportError, ModuleNotFoundError): |
| 155 | from sanitize_repo_content import _escape_untrusted_boundaries |
| 156 | safe_wisdom = ( |
| 157 | _escape_untrusted_boundaries(wisdom_content) |
| 158 | if wisdom_content |
| 159 | else "(No per-topic wisdom accumulated yet.)" |
| 160 | ) |
| 161 | rendered = rendered.replace("{{WISDOM_CONTENT}}", safe_wisdom) |
| 162 | else: |
| 163 | # Remove IF_TOPIC blocks |
| 164 | rendered = _remove_blocks(template, "IF_TOPIC") |
| 165 | # Keep IF_NO_TOPIC block contents |
| 166 | rendered = _keep_blocks(rendered, "IF_NO_TOPIC") |
| 167 | |
| 168 | # Clear any remaining topic placeholders |
| 169 | rendered = rendered.replace("{{TOPIC_ID}}", "") |
| 170 | rendered = rendered.replace("{{TOPIC_NAME}}", "") |
| 171 | rendered = rendered.replace("{{TOPIC_DESCRIPTION}}", "") |
| 172 | rendered = rendered.replace("{{WISDOM_CONTENT}}", "") |
| 173 | |
| 174 | return rendered |
| 175 | |
| 176 | |
| 177 | def _remove_blocks(text: str, block_name: str) -> str: |
| 178 | """Remove conditional block markers and their contents.""" |
| 179 | start_tag = "{{#" + block_name + "}}" |
| 180 | end_tag = "{{/" + block_name + "}}" |
| 181 | |
| 182 | result = text |
| 183 | while start_tag in result: |
| 184 | start_idx = result.index(start_tag) |
| 185 | end_idx = result.index(end_tag) + len(end_tag) |
| 186 | # Remove trailing newline if present |
| 187 | if end_idx < len(result) and result[end_idx] == "\n": |
| 188 | end_idx += 1 |
| 189 | result = result[:start_idx] + result[end_idx:] |
| 190 | |
| 191 | return result |
| 192 | |
| 193 | |
| 194 | def _keep_blocks(text: str, block_name: str) -> str: |
| 195 | """Remove conditional block markers but keep their contents.""" |
| 196 | start_tag = "{{#" + block_name + "}}" |
| 197 | end_tag = "{{/" + block_name + "}}" |
| 198 | |
| 199 | result = text.replace(start_tag + "\n", "").replace(start_tag, "") |
| 200 | result = result.replace(end_tag + "\n", "").replace(end_tag, "") |
| 201 | return result |
| 202 | |
| 203 | |
| 204 | def main() -> None: |
| 205 | parser = argparse.ArgumentParser(description="Render topic-aware analysis prompt template") |
| 206 | parser.add_argument("--config", type=Path, help="Path to squadscope.topic.yml") |
| 207 | parser.add_argument("--output", type=Path, help="Output file path (default: stdout)") |
| 208 | parser.add_argument( |
| 209 | "--template", type=Path, help="Template file (default: prompts/analyze-topic.md)" |
| 210 | ) |
| 211 | args = parser.parse_args() |
| 212 | |
| 213 | root = find_repo_root() |
| 214 | |
| 215 | # Load template |
| 216 | template_path = args.template or (root / "prompts" / "analyze-topic.md") |
| 217 | if not template_path.exists(): |
| 218 | print(f"Error: Template not found at {template_path}", file=sys.stderr) |
| 219 | sys.exit(1) |
| 220 | template = template_path.read_text(encoding="utf-8") |
| 221 | |
| 222 | # Load topic config |
| 223 | topic_config = load_topic_config(args.config) |
| 224 | |
| 225 | # Render |
| 226 | rendered = render_template(template, topic_config) |
| 227 | |
| 228 | # Output |
| 229 | if args.output: |
| 230 | args.output.parent.mkdir(parents=True, exist_ok=True) |
| 231 | args.output.write_text(rendered, encoding="utf-8") |
| 232 | else: |
| 233 | print(rendered) |
| 234 | |
| 235 | |
| 236 | if __name__ == "__main__": |
| 237 | main() |