1"""Shared OpenSpec spec parser and semantic diff.23Importable by both ``tools.py`` (agent tool handlers) and4``dashboard/plugin_api.py`` (dashboard backend) so the spec parsing5and diffing logic lives in one place.67The parser mirrors the frontend ``parseSpec`` function in8``dashboard/dist/index.js`` — same state machine, same output structure.9"""1011from __future__ import annotations1213import difflib14from typing import Any151617def parse_spec(md: str | None) -> dict[str, Any]:18 """Parse OpenSpec spec markdown into structured data.1920 Returns ``{ title, purpose, requirements: [...] }`` where each21 requirement is ``{ name, description, scenarios: [...] }`` and each22 scenario is ``{ name, steps: [{ type, text }] }``.23 """24 if not md:25 return {"title": "", "purpose": "", "requirements": []}2627 lines = md.split("\n")28 title = ""29 purpose = ""30 reqs: list[dict[str, Any]] = []31 state = "top" # top | purpose | reqDesc | scenario32 current_req: dict[str, Any] | None = None33 current_scn: dict[str, Any] | None = None3435 for line in lines:36 trim = line.strip()3738 # Title: "# Title" but not "## Title"39 if trim.startswith("# ") and not trim.startswith("## "):40 title = trim[2:].strip().removesuffix("Specification").strip()41 state = "top"42 continue4344 if trim.lower().startswith("## purpose"):45 state = "purpose"46 continue4748 if trim.lower().startswith("## requirements"):49 state = "top"50 continue5152 m_req = re_match(r"^###\s+Requirement:\s*(.+)$", trim)53 if m_req:54 current_req = {"name": m_req.group(1).strip(), "description": "", "scenarios": []}55 reqs.append(current_req)56 current_scn = None57 state = "reqDesc"58 continue5960 m_scn = re_match(r"^####\s+Scenario:\s*(.+)$", trim)61 if m_scn:62 current_scn = {"name": m_scn.group(1).strip(), "steps": []}63 if current_req is not None:64 current_req["scenarios"].append(current_scn)65 state = "scenario"66 continue6768 # Accumulate text based on state69 if not trim:70 continue7172 if state == "purpose":73 purpose = f"{purpose}\n{trim}" if purpose else trim74 elif state == "reqDesc":75 if current_req is not None:76 current_req["description"] = (77 f"{current_req['description']}\n{trim}"78 if current_req["description"]79 else trim80 )81 elif state == "scenario":82 if current_scn is not None:83 step = re_match(r"^[-*]\s+\*\*(\w+)\*\*\s*(.+)$", trim)84 if step:85 current_scn["steps"].append({86 "type": step.group(1).upper(),87 "text": step.group(2).strip(),88 })89 elif current_scn["steps"]:90 # Non-bullet text inside scenario — append to last step91 current_scn["steps"][-1]["text"] += f" {trim}"9293 return {"title": title, "purpose": purpose, "requirements": reqs}949596def _scenario_key(scn: dict[str, Any]) -> str:97 """Normalize a scenario for comparison by name."""98 return scn["name"]99100101def _requirement_key(req: dict[str, Any]) -> str:102 """Normalize a requirement for comparison by name."""103 return req["name"]104105106def _scenarios_equal(a: dict[str, Any] | None, b: dict[str, Any] | None) -> bool:107 if a is None or b is None:108 return a is b109 return a["name"] == b["name"] and a["steps"] == b["steps"]110111112def _requirements_equal(a: dict[str, Any], b: dict[str, Any]) -> bool:113 return (114 a["name"] == b["name"]115 and a["description"] == b["description"]116 and a["scenarios"] == b["scenarios"]117 )118119120def semantic_spec_diff(before_md: str | None, after_md: str | None) -> dict[str, Any]:121 """Compare two spec markdown strings at the requirement/scenario level.122123 Returns a structured delta::124125 {126 "status": "added" | "modified" | "deleted" | "unchanged",127 "requirements": {128 "added": [...],129 "modified": [{ name, before, after, scenarios_added, ... }],130 "removed": [...],131 "unchanged": ["name", ...],132 },133 }134 """135 before = parse_spec(before_md)136 after = parse_spec(after_md)137138 before_reqs = {r["name"]: r for r in before["requirements"]}139 after_reqs = {r["name"]: r for r in after["requirements"]}140141 added: list[dict[str, Any]] = []142 modified: list[dict[str, Any]] = []143 removed: list[dict[str, Any]] = []144 unchanged: list[str] = []145146 all_names = set(before_reqs) | set(after_reqs)147 for name in sorted(all_names):148 b_req = before_reqs.get(name)149 a_req = after_reqs.get(name)150151 if b_req is None and a_req is not None:152 added.append(a_req)153 elif b_req is not None and a_req is None:154 removed.append(b_req)155 elif b_req is not None and a_req is not None:156 if _requirements_equal(b_req, a_req):157 unchanged.append(name)158 else:159 # Diff scenarios within the modified requirement160 b_scns = {s["name"]: s for s in b_req["scenarios"]}161 a_scns = {s["name"]: s for s in a_req["scenarios"]}162163 scn_added: list[dict[str, Any]] = []164 scn_modified: list[dict[str, Any]] = []165 scn_removed: list[dict[str, Any]] = []166167 for scn_name in sorted(set(b_scns) | set(a_scns)):168 b_scn = b_scns.get(scn_name)169 a_scn = a_scns.get(scn_name)170 if b_scn is None and a_scn is not None:171 scn_added.append(a_scn)172 elif b_scn is not None and a_scn is None:173 scn_removed.append(b_scn)174 elif b_scn is not None and a_scn is not None:175 if not _scenarios_equal(b_scn, a_scn):176 scn_modified.append({177 "name": scn_name,178 "before": b_scn,179 "after": a_scn,180 })181182 modified.append({183 "name": name,184 "before": {185 "description": b_req["description"],186 "scenarios": b_req["scenarios"],187 },188 "after": {189 "description": a_req["description"],190 "scenarios": a_req["scenarios"],191 },192 "scenarios_added": scn_added,193 "scenarios_modified": scn_modified,194 "scenarios_removed": scn_removed,195 })196197 # Overall status198 if not before_md and not after_md:199 status = "unchanged"200 elif not before_md:201 status = "added"202 elif not after_md:203 status = "deleted"204 elif not added and not modified and not removed:205 status = "unchanged"206 else:207 status = "modified"208209 return {210 "status": status,211 "requirements": {212 "added": added,213 "modified": modified,214 "removed": removed,215 "unchanged": unchanged,216 },217 }218219220def semantic_summary(diff: dict[str, Any]) -> dict[str, int]:221 """Extract compact counts from a semantic diff for list display."""222 reqs = diff.get("requirements", {})223 return {224 "added": len(reqs.get("added", [])),225 "modified": len(reqs.get("modified", [])),226 "removed": len(reqs.get("removed", [])),227 }228229230def unified_diff(before_md: str | None, after_md: str | None, path: str = "spec") -> str:231 """Return a unified line diff for fallback display."""232 return "\n".join(233 difflib.unified_diff(234 (before_md or "").splitlines(),235 (after_md or "").splitlines(),236 fromfile=f"before/{path}",237 tofile=f"after/{path}",238 lineterm="",239 )240 )241242243def spec_to_markdown(title: str, purpose: str, requirements: list[dict[str, Any]]) -> str:244 """Serialize structured spec data into OpenSpec spec markdown.245246 Inverse of ``parse_spec`` — takes ``{title, purpose, requirements}`` and247 produces ``# Title`` / ``## Purpose`` / ``## Requirements`` /248 ``### Requirement:`` / ``#### Scenario:`` formatted markdown.249250 Each requirement is ``{name, description, scenarios}`` and each251 scenario is ``{name, steps: [{type, text}]}``.252 """253 lines: list[str] = [f"# {title}", ""]254255 if purpose:256 lines.append("## Purpose")257 lines.append("")258 lines.append(purpose)259 lines.append("")260261 if requirements:262 lines.append("## Requirements")263 lines.append("")264 for req in requirements:265 name = str(req.get("name") or "").strip()266 desc = str(req.get("description") or "").strip()267 lines.append(f"### Requirement: {name}")268 if desc:269 lines.append(desc)270 lines.append("")271 for scn in req.get("scenarios") or []:272 scn_name = str(scn.get("name") or "").strip()273 lines.append(f"#### Scenario: {scn_name}")274 lines.append("")275 for step in scn.get("steps") or []:276 step_type = str(step.get("type") or "THEN").strip().upper()277 step_text = str(step.get("text") or "").strip()278 lines.append(f"- **{step_type}** {step_text}")279 lines.append("")280281 # Trim trailing blank lines, ensure single trailing newline282 return "\n".join(lines).rstrip() + "\n"283284285def re_match(pattern: str, string: str):286 """Wrapper to avoid importing re at module level in callers."""287 import re288 return re.match(pattern, string, re.IGNORECASE)