1"""OpenSpec Hermes tool handlers."""23from __future__ import annotations45import json6import os7import re8import shutil9import subprocess10from datetime import datetime, timezone11from pathlib import Path12from typing import Any131415MAX_OUTPUT_CHARS = 50_0001617_ARTIFACT_ALIASES = {"spec": "specs"}181920def _openspec_bin() -> str | None:21 configured = os.getenv("OPENSPEC_BIN", "").strip()22 candidates = [23 configured,24 shutil.which("openspec") or "",25 str(Path.home() / ".npm-global" / "bin" / "openspec"),26 ]27 for candidate in candidates:28 if candidate and Path(candidate).exists() and os.access(candidate, os.X_OK):29 return candidate30 return None313233def _resolve_workdir(value: Any) -> tuple[Path | None, str | None]:34 raw = str(value or "").strip()35 if not raw:36 return Path.cwd(), None37 p = Path(raw).expanduser()38 if not p.is_absolute():39 p = (Path.cwd() / p).resolve()40 if not p.exists() or not p.is_dir():41 return None, f"workdir does not exist or is not a directory: {p}"42 return p, None434445def _json_or_text(text: str) -> Any:46 stripped = text.strip()47 if not stripped:48 return ""49 try:50 return json.loads(stripped)51 except Exception:52 return stripped535455def _normalize_artifact(value: str) -> str:56 artifact = (value or "").strip()57 return _ARTIFACT_ALIASES.get(artifact, artifact)585960def _run(args: list[str], workdir: Any = None) -> str:61 exe = _openspec_bin()62 if not exe:63 return json.dumps({64 "ok": False,65 "error": "openspec executable not found. Install OpenSpec or set OPENSPEC_BIN.",66 })6768 cwd, err = _resolve_workdir(workdir)69 if err:70 return json.dumps({"ok": False, "error": err})7172 env = os.environ.copy()73 env.setdefault("OPENSPEC_TELEMETRY", "0")74 env.setdefault("NO_COLOR", "1")7576 cmd = [exe, *args]77 try:78 proc = subprocess.run(79 cmd,80 cwd=str(cwd),81 env=env,82 text=True,83 stdout=subprocess.PIPE,84 stderr=subprocess.PIPE,85 timeout=120,86 check=False,87 )88 except subprocess.TimeoutExpired:89 return json.dumps({90 "ok": False,91 "command": cmd,92 "workdir": str(cwd),93 "error": "openspec command timed out after 120s",94 })95 except Exception as exc:96 return json.dumps({97 "ok": False,98 "command": cmd,99 "workdir": str(cwd),100 "error": str(exc),101 })102103 stdout = proc.stdout[-MAX_OUTPUT_CHARS:]104 stderr = proc.stderr[-MAX_OUTPUT_CHARS:]105 return json.dumps({106 "ok": proc.returncode == 0,107 "exit_code": proc.returncode,108 "command": cmd,109 "workdir": str(cwd),110 "stdout": _json_or_text(stdout),111 "stderr": _json_or_text(stderr),112 "truncated": len(proc.stdout) > MAX_OUTPUT_CHARS or len(proc.stderr) > MAX_OUTPUT_CHARS,113 })114115116def _template_instruction_fallback(artifact: str, reason: str, workdir: Any = None, schema: str = "") -> str | None:117 """Return template-backed instructions when OpenSpec needs a change first."""118 if "No changes found" not in reason:119 return None120 artifact = _normalize_artifact(artifact)121 if not artifact:122 return None123124 templates_args = ["templates", "--json"]125 if schema:126 templates_args.extend(["--schema", schema])127 templates_result = json.loads(_run(templates_args, workdir))128 if not templates_result.get("ok") or not isinstance(templates_result.get("stdout"), dict):129 return None130131 entry = templates_result["stdout"].get(artifact)132 if not isinstance(entry, dict):133 return None134 path = Path(str(entry.get("path") or "")).expanduser()135 if not path.is_file():136 return None137 try:138 content = path.read_text(encoding="utf-8", errors="replace")139 except OSError:140 return None141142 fallback = dict(templates_result)143 fallback["ok"] = True144 fallback["stdout"] = {145 "artifact": artifact,146 "fallback": "template",147 "reason": reason,148 "templatePath": str(path),149 "source": entry.get("source"),150 "content": content,151 }152 fallback["stderr"] = ""153 return json.dumps(fallback)154155156def _registry_module():157 """Import the plugin-local OpenSpec registry (``plugins.openspec.registry``).158159 Tolerates odd sys.path setups where the package isn't importable by name;160 in that case fall back to a path-based load from this file's location.161 """162 try:163 from . import registry # type: ignore164 return registry165 except Exception:166 pass167 try:168 import importlib.util169 candidate = Path(__file__).resolve().parent / "registry.py"170 if candidate.is_file():171 spec = importlib.util.spec_from_file_location("openspec_registry", candidate)172 if spec and spec.loader:173 mod = importlib.util.module_from_spec(spec)174 spec.loader.exec_module(mod)175 return mod176 except Exception:177 pass178 return None179180181def _read_doc(path: Path) -> str:182 try:183 if path.is_file():184 return path.read_text(encoding="utf-8", errors="replace")185 except OSError:186 pass187 return ""188189190def _iter_change_dirs(root: Path):191 """Yield ``(change_dir, archived)`` for every change folder with a proposal."""192 changes_root = root / "openspec" / "changes"193 if changes_root.is_dir():194 for child in sorted(changes_root.iterdir()):195 if child.name == "archive":196 continue197 if child.is_dir() and (child / "proposal.md").is_file():198 yield child, False199 archive_root = changes_root / "archive"200 if archive_root.is_dir():201 for child in sorted(archive_root.iterdir()):202 if child.is_dir() and (child / "proposal.md").is_file():203 yield child, True204205206def _resolve_change(root: Path, registry, change_ref: str) -> tuple[Path | None, bool]:207 """Map a change token (``os_xxx``) — or a literal change name (legacy) — to a208 change folder under ``root``. Returns ``(change_dir, archived)``."""209 for change_dir, archived in _iter_change_dirs(root):210 if registry.change_token(change_dir.name) == change_ref or change_dir.name == change_ref:211 return change_dir, archived212 return None, False213214215def _iter_specs(root: Path):216 specs_root = root / "openspec" / "specs"217 if specs_root.is_dir():218 for child in sorted(specs_root.rglob("*.md")):219 if child.is_file():220 yield child.relative_to(specs_root).as_posix(), child221222223def _resolve_spec(root: Path, registry, spec_ref: str) -> tuple[str | None, Path | None]:224 for rel, path in _iter_specs(root):225 if registry.change_token(f"spec:{rel}") == spec_ref or rel == spec_ref:226 return rel, path227 return None, None228229230def _iter_ideas(root: Path):231 """Yield every ``.md`` idea file under ``openspec/ideas/`` (sorted)."""232 ideas_root = root / "openspec" / "ideas"233 if ideas_root.is_dir():234 for child in sorted(ideas_root.iterdir()):235 if child.is_file() and child.suffix.lower() == ".md":236 yield child237238239def _resolve_idea(root: Path, registry, idea_ref: str) -> Path | None:240 """Map an idea token (``os_xxx``) — or a literal idea stem (legacy) — to the241 matching idea file. Returns ``None`` when nothing matches."""242 for idea in _iter_ideas(root):243 if registry.change_token(idea.stem) == idea_ref or idea.stem == idea_ref:244 return idea245 return None246247248def openspec_context(args: dict, **kwargs) -> str:249 """Resolve a copyable identifier (``puzzletea`` or ``puzzletea/os_xxx``) into250 repo path + (optionally) the change's proposal/tasks/design/specs content."""251 registry = _registry_module()252 if registry is None:253 return json.dumps({254 "ok": False,255 "error": "OpenSpec registry unavailable (plugins.openspec.registry not importable).",256 })257258 identifier = str(args.get("identifier") or "").strip()259 if not identifier:260 return json.dumps({"ok": False, "error": "identifier is required (e.g. 'puzzletea' or 'puzzletea/os_a1b2c3')"})261262 source_ref, change_ref = registry.parse_identifier(identifier)263 # Resolve source by vanity name first, then fall back to a legacy source token.264 source = registry.get_source_by_name(source_ref) or registry.get_source(source_ref)265 if source is None:266 return json.dumps({"ok": False, "error": f"No registered OpenSpec source for '{source_ref}'."})267268 name = registry.effective_name(source)269 repo_path = Path(source["path"]).expanduser()270 if not (repo_path / "openspec").is_dir():271 return json.dumps({272 "ok": False,273 "name": name,274 "path": str(repo_path),275 "error": f"openspec/ directory not found at {repo_path} (repo may have moved).",276 })277278 result: dict[str, Any] = {279 "ok": True,280 "name": name,281 "path": str(repo_path),282 "workdir": str(repo_path),283 "hint": "Use this path as 'workdir' for other openspec_* tools, or run openspec commands from here.",284 }285286 if change_ref:287 change_dir, archived = _resolve_change(repo_path, registry, change_ref)288 spec_rel, spec_path = _resolve_spec(repo_path, registry, change_ref)289 idea_path = _resolve_idea(repo_path, registry, change_ref)290 matches = sum(291 1 for matched in (change_dir is not None, spec_rel is not None, idea_path is not None) if matched292 )293 if matches > 1:294 return json.dumps({295 "ok": False,296 "name": name,297 "path": str(repo_path),298 "error": (299 f"Reference '{change_ref}' is ambiguous: it matches more than one of the supported "300 "artifact kinds (change, spec, idea). Use a literal change name, spec path, or idea "301 "name instead."302 ),303 })304 if idea_path is not None:305 result["idea"] = {306 "name": idea_path.stem,307 "filename": idea_path.name,308 "token": registry.change_token(idea_path.stem),309 "content": _read_doc(idea_path),310 }311 return json.dumps(result)312 if change_dir is None:313 if spec_rel and spec_path:314 result["spec"] = {315 "path": spec_rel,316 "token": registry.change_token(f"spec:{spec_rel}"),317 "content": _read_doc(spec_path),318 }319 return json.dumps(result)320 return json.dumps({321 "ok": False,322 "name": name,323 "path": str(repo_path),324 "error": f"No change, spec, or idea matching '{change_ref}' under openspec/.",325 })326 specs_root = change_dir / "specs"327 specs = []328 if specs_root.is_dir():329 for child in sorted(specs_root.rglob("*.md")):330 specs.append({331 "path": child.relative_to(specs_root).as_posix(),332 "content": _read_doc(child),333 })334 result["change"] = {335 "name": change_dir.name,336 "token": registry.change_token(change_dir.name),337 "archived": archived,338 "proposal": _read_doc(change_dir / "proposal.md"),339 "tasks": _read_doc(change_dir / "tasks.md"),340 "design": _read_doc(change_dir / "design.md"),341 "specs": specs,342 }343 else:344 change_dirs = [(change_dir, archived) for change_dir, archived in _iter_change_dirs(repo_path) if not archived]345 sequence = {}346 if hasattr(registry, "ensure_change_sequence"):347 sequence = registry.ensure_change_sequence(str(source.get("id") or source.get("token")), [change_dir.name for change_dir, _ in change_dirs])348 changes = [349 {350 "name": change_dir.name,351 "token": registry.change_token(change_dir.name),352 **({"sequence": sequence[change_dir.name]} if change_dir.name in sequence else {}),353 }354 for change_dir, archived in change_dirs355 ]356 specs = [357 {"path": rel, "token": registry.change_token(f"spec:{rel}")}358 for rel, _path in _iter_specs(repo_path)359 ]360 ideas = [361 {"name": idea.stem, "filename": idea.name, "token": registry.change_token(idea.stem)}362 for idea in _iter_ideas(repo_path)363 ]364 result["changes"] = changes365 result["specs"] = specs366 result["ideas"] = ideas367 result["hint"] = (368 "Active changes are listed in 'changes'; current specs are listed in 'specs'; ideas are "369 "listed in 'ideas'. Re-call with identifier "370 f"'{name}/<token>' to load a specific change, spec, or idea, or use workdir with other "371 "openspec_* tools."372 )373374 return json.dumps(result)375376377_SLUG_RE = re.compile(r"[^a-z0-9]+")378_VALID_CHANGE_RE = re.compile(r"^[a-z0-9][a-z0-9-]*[a-z0-9]$|^[a-z0-9]$")379_ENRICHMENT_START = "<!-- OPENSPEC_IDEA_ENRICHMENT_START -->"380_ENRICHMENT_END = "<!-- OPENSPEC_IDEA_ENRICHMENT_END -->"381_VALID_FEASIBILITY = {"low", "medium", "high"}382_VALID_TSHIRT = {"xs", "s", "m", "l", "xl"}383384385def _error(message: str, **extra: Any) -> str:386 payload = {"ok": False, "error": message}387 payload.update(extra)388 return json.dumps(payload)389390391def _now_iso() -> str:392 return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")393394395def _slugify(value: str, fallback: str = "idea") -> str:396 slug = _SLUG_RE.sub("-", value.strip().lower()).strip("-")397 return slug or fallback398399400def _resolve_project(args: dict) -> tuple[Path | None, str | None, dict[str, Any]]:401 """Resolve a tool call to a project root.402403 Prefer explicit workdir. Also accept a registered OpenSpec identifier/source404 name so agents can call write tools with the same project handles used by405 ``openspec_context``.406 """407 workdir = args.get("workdir")408 if workdir:409 root, err = _resolve_workdir(workdir)410 return root, err, {}411412 identifier = str(args.get("identifier") or args.get("project") or "").strip()413 if identifier:414 registry = _registry_module()415 if registry is None:416 return None, "OpenSpec registry unavailable (plugins.openspec.registry not importable).", {}417 source_ref, _artifact_ref = registry.parse_identifier(identifier)418 source = registry.get_source_by_name(source_ref) or registry.get_source(source_ref)419 if source is None:420 return None, f"No registered OpenSpec source for '{source_ref}'.", {}421 root = Path(source["path"]).expanduser()422 if not root.is_absolute():423 root = root.resolve()424 if not root.exists() or not root.is_dir():425 return None, f"registered source path does not exist or is not a directory: {root}", {}426 return root, None, {"source": source, "name": registry.effective_name(source)}427428 root, err = _resolve_workdir(None)429 return root, err, {}430431432def _ensure_openspec_layout(root: Path) -> None:433 openspec_root = root / "openspec"434 (openspec_root / "changes" / "archive").mkdir(parents=True, exist_ok=True)435 (openspec_root / "specs").mkdir(parents=True, exist_ok=True)436 (openspec_root / "ideas").mkdir(parents=True, exist_ok=True)437438439def _unique_idea_path(ideas_root: Path, slug: str) -> tuple[str, Path]:440 candidate = ideas_root / f"{slug}.md"441 if not candidate.exists():442 return slug, candidate443 index = 2444 while True:445 suffixed = f"{slug}-{index}"446 candidate = ideas_root / f"{suffixed}.md"447 if not candidate.exists():448 return suffixed, candidate449 index += 1450451452def _coerce_string_list(value: Any) -> list[str]:453 if value is None or value == "":454 return []455 if isinstance(value, str):456 return [line.strip() for line in value.splitlines() if line.strip()]457 if isinstance(value, list):458 return [str(item).strip() for item in value if str(item).strip()]459 return [str(value).strip()]460461462def _idea_path(root: Path, idea_ref: str) -> Path:463 slug = _slugify(idea_ref)464 return root / "openspec" / "ideas" / f"{slug}.md"465466467def _relative(path: Path, root: Path) -> str:468 try:469 return path.relative_to(root).as_posix()470 except ValueError:471 return path.as_posix()472473474def _markdown_list(items: list[str]) -> str:475 if not items:476 return "- None"477 return "\n".join(f"- {item}" for item in items)478479480_TASK_LINE_RE = re.compile(r"^(?P<prefix>\s*- \[)(?P<mark>[ xX])(?P<suffix>\]\s*(?:(?P<id>\d+(?:\.\d+)*)\s+)?(?P<text>.*?))\s*$")481482483def _change_path(root: Path, change: str, *, archived: bool = False) -> Path:484 base = root / "openspec" / "changes"485 if archived:486 base = base / "archive"487 return base / change488489490def _resolve_change_path(root: Path, change_ref: str) -> tuple[Path | None, bool]:491 change = _slugify(change_ref, fallback="")492 if not change:493 return None, False494 active = _change_path(root, change)495 if active.is_dir():496 return active, False497 archived = _change_path(root, change, archived=True)498 if archived.is_dir():499 return archived, True500 return None, False501502503def _resolve_change_names(root: Path, registry, refs: list[str]) -> tuple[list[str], list[str]]:504 names: list[str] = []505 missing: list[str] = []506 by_name = {change_dir.name: change_dir.name for change_dir, _ in _iter_change_dirs(root)}507 by_token = {registry.change_token(name): name for name in by_name}508 for ref in refs:509 raw = str(ref or "").strip()510 if not raw:511 continue512 name = by_name.get(raw) or by_token.get(raw)513 if name:514 if name not in names:515 names.append(name)516 else:517 missing.append(raw)518 return names, missing519520521def _derive_status(tasks_path: Path, *, archived: bool = False) -> str:522 if archived:523 return "archived"524 if not tasks_path.is_file():525 return "draft"526 tasks = _parse_tasks(tasks_path)527 if not tasks:528 return "draft"529 done = sum(1 for task in tasks if task["status"] == "done")530 if done == 0:531 return "todo"532 if done == len(tasks):533 return "done"534 return "in_progress"535536537def _parse_tasks(tasks_path: Path) -> list[dict[str, Any]]:538 if not tasks_path.is_file():539 return []540 tasks: list[dict[str, Any]] = []541 seq = 0542 for line_no, line in enumerate(tasks_path.read_text(encoding="utf-8", errors="replace").splitlines(), start=1):543 match = _TASK_LINE_RE.match(line)544 if not match:545 continue546 seq += 1547 task_id = match.group("id") or str(seq)548 tasks.append({549 "id": task_id,550 "text": match.group("text").strip(),551 "status": "done" if match.group("mark").lower() == "x" else "todo",552 "line": line_no,553 })554 return tasks555556557def _task_counts(tasks: list[dict[str, Any]]) -> dict[str, int]:558 done = sum(1 for task in tasks if task["status"] == "done")559 total = len(tasks)560 return {"total": total, "done": done, "todo": total - done}561562563def _format_tasks(items: list[str]) -> str:564 lines = ["## 1. Tasks", ""]565 for index, text in enumerate(items, start=1):566 lines.append(f"- [ ] 1.{index} {text}")567 lines.append("")568 return "\n".join(lines)569570571def _default_tasks() -> list[str]:572 return ["Refine scope", "Implement behavior", "Validate OpenSpec artifacts and tests"]573574575def _write_placeholder_spec(change_dir: Path, change: str, title: str, source: str = "change") -> Path:576 spec_dir = change_dir / "specs" / change577 spec_dir.mkdir(parents=True, exist_ok=True)578 spec_path = spec_dir / "spec.md"579 if spec_path.exists():580 return spec_path581 spec = f"""# {title}582583## ADDED Requirements584585### Requirement: {title} scope586The system SHALL implement the approved behavior described by this {source} after this placeholder is refined into concrete requirements.587588#### Scenario: Placeholder refinement589- **GIVEN** `{change}` has been created as an OpenSpec change590- **WHEN** implementation begins591- **THEN** this placeholder requirement is replaced or refined with concrete, testable behavior before the change is completed592"""593 spec_path.write_text(spec, encoding="utf-8")594 return spec_path595596597def _write_change_scaffold(598 root: Path,599 change: str,600 title: str,601 summary: str,602 *,603 source_section: str = "",604 tasks: list[str] | None = None,605 with_spec: bool = False,606 force: bool = False,607) -> tuple[dict[str, Any] | None, str | None]:608 _ensure_openspec_layout(root)609 change_dir = _change_path(root, change)610 if change_dir.exists() and not force:611 return None, f"change already exists: {change}"612 change_dir.mkdir(parents=True, exist_ok=True)613 created = _now_iso()614 source_block = f"\n## Source\n\n{source_section}\n" if source_section else ""615 proposal = f"""## Why616617{summary}618619## What Changes620621- Create the `{change}` OpenSpec change scaffold for review and refinement.622623{source_block}## Capabilities624625### New Capabilities626- `{change}`: Initial capability placeholder. Refine before implementation.627628### Modified Capabilities629- None yet.630631## Impact632633- TBD: Fill in affected code, APIs, dependencies, or systems before implementation.634"""635 (change_dir / "proposal.md").write_text(proposal, encoding="utf-8")636 task_items = tasks or []637 if task_items:638 (change_dir / "tasks.md").write_text(_format_tasks(task_items), encoding="utf-8")639 spec_path = None640 if with_spec:641 spec_path = _write_placeholder_spec(change_dir, change, title)642 (change_dir / ".openspec.yaml").write_text(643 "schema: spec-driven\ncreatedAt: " + json.dumps(created) + "\n",644 encoding="utf-8",645 )646 paths = {647 "proposal": str(change_dir / "proposal.md"),648 "metadata": str(change_dir / ".openspec.yaml"),649 }650 if task_items:651 paths["tasks"] = str(change_dir / "tasks.md")652 if spec_path:653 paths["spec"] = str(spec_path)654 return {655 "ok": True,656 "change": change,657 "change_path": str(change_dir),658 "relative_change_path": _relative(change_dir, root),659 "workdir": str(root),660 "status": _derive_status(change_dir / "tasks.md"),661 "paths": paths,662 }, None663664665def openspec_idea_create(args: dict, **kwargs) -> str:666 root, err, context = _resolve_project(args)667 if err or root is None:668 return _error(err or "workdir could not be resolved")669670 title = str(args.get("title") or "").strip()671 prompt = str(args.get("prompt") or "").strip()672 if not title:673 return _error("title is required")674 if not prompt:675 return _error("prompt is required")676677 origin = str(args.get("origin") or args.get("source") or "unspecified").strip() or "unspecified"678 notes = str(args.get("notes") or "").strip()679 tags = _coerce_string_list(args.get("tags"))680681 _ensure_openspec_layout(root)682 ideas_root = root / "openspec" / "ideas"683 slug, path = _unique_idea_path(ideas_root, _slugify(str(args.get("slug") or title)))684 created = _now_iso()685 tags_text = ", ".join(tags) if tags else "None"686687 sections = [688 f"# {title}",689 "",690 "## Source",691 f"- Origin: {origin}",692 f"- Created: {created}",693 f"- Tags: {tags_text}",694 ]695 if context.get("name"):696 sections.append(f"- Project: {context['name']}")697 sections.extend([698 "",699 "## Prompt",700 prompt,701 ])702 if notes:703 sections.extend(["", "## Notes", notes])704 sections.append("")705706 path.write_text("\n".join(sections), encoding="utf-8")707 return json.dumps({708 "ok": True,709 "slug": slug,710 "title": title,711 "path": str(path),712 "relative_path": _relative(path, root),713 "workdir": str(root),714 "metadata": {"origin": origin, "created": created, "tags": tags},715 })716717718def openspec_idea_enrich(args: dict, **kwargs) -> str:719 root, err, _context = _resolve_project(args)720 if err or root is None:721 return _error(err or "workdir could not be resolved")722 idea = str(args.get("idea") or args.get("slug") or "").strip()723 if not idea:724 return _error("idea is required")725726 path = _idea_path(root, idea)727 if not path.is_file():728 return _error(f"idea not found: {idea}")729730 feasibility = str(args.get("feasibility") or "").strip()731 tshirt_size = str(args.get("tshirt_size") or args.get("t_shirt_size") or "").strip()732 if feasibility and feasibility.lower() not in _VALID_FEASIBILITY:733 return _error("feasibility must be one of: Low, Medium, High")734 if tshirt_size and tshirt_size.lower() not in _VALID_TSHIRT:735 return _error("tshirt_size must be one of: XS, S, M, L, XL")736737 problem = str(args.get("problem") or "TBD").strip() or "TBD"738 proposed_direction = str(args.get("proposed_direction") or args.get("proposedDirection") or "TBD").strip() or "TBD"739 key_questions = _coerce_string_list(args.get("key_questions"))740 risks = _coerce_string_list(args.get("risks"))741 size_justification = str(args.get("size_justification") or "TBD").strip() or "TBD"742 suggested_next_step = str(args.get("suggested_next_step") or "TBD").strip() or "TBD"743 generated = _now_iso()744745 report = "\n".join([746 _ENRICHMENT_START,747 "## Enrichment Report",748 "",749 f"Generated: {generated}",750 "",751 "### Problem",752 problem,753 "",754 "### Proposed Direction",755 proposed_direction,756 "",757 "### Key Questions",758 _markdown_list(key_questions),759 "",760 "### Feasibility",761 f"Feasibility: {feasibility or 'TBD'}",762 "",763 "### T-Shirt Size",764 f"T-Shirt Size: {tshirt_size or 'TBD'}",765 "",766 "### Size Justification",767 size_justification,768 "",769 "### Risks",770 _markdown_list(risks),771 "",772 "### Suggested Next Step",773 suggested_next_step,774 _ENRICHMENT_END,775 "",776 ])777778 content = path.read_text(encoding="utf-8", errors="replace")779 if _ENRICHMENT_START in content and _ENRICHMENT_END in content:780 before, rest = content.split(_ENRICHMENT_START, 1)781 _old, after = rest.split(_ENRICHMENT_END, 1)782 content = before.rstrip() + "\n\n" + report + after.lstrip("\n")783 else:784 content = content.rstrip() + "\n\n" + report785 path.write_text(content, encoding="utf-8")786787 return json.dumps({788 "ok": True,789 "idea": _slugify(idea),790 "path": str(path),791 "relative_path": _relative(path, root),792 "workdir": str(root),793 "metadata": {"generated": generated, "feasibility": feasibility, "tshirt_size": tshirt_size},794 })795796797def openspec_idea_promote(args: dict, **kwargs) -> str:798 root, err, _context = _resolve_project(args)799 if err or root is None:800 return _error(err or "workdir could not be resolved")801 idea = str(args.get("idea") or args.get("slug") or "").strip()802 change = _slugify(str(args.get("change") or args.get("change_id") or ""), fallback="")803 if not idea:804 return _error("idea is required")805 if not change or not _VALID_CHANGE_RE.match(change):806 return _error("change must be a kebab-case id")807808 idea_path = _idea_path(root, idea)809 if not idea_path.is_file():810 return _error(f"idea not found: {idea}")811812 _ensure_openspec_layout(root)813 change_dir = root / "openspec" / "changes" / change814 if change_dir.exists() and not args.get("force"):815 return _error(f"change already exists: {change}", change=change, change_path=str(change_dir))816 change_dir.mkdir(parents=True, exist_ok=True)817818 idea_content = idea_path.read_text(encoding="utf-8", errors="replace")819 title = idea_content.splitlines()[0].lstrip("# ").strip() if idea_content.strip() else idea820 summary = str(args.get("summary") or f"Promote idea: {title}").strip()821 rel_idea = _relative(idea_path, root)822 created = _now_iso()823824 proposal = f"""## Why825826{summary}827828## What Changes829830- Promote the reviewed idea `{idea_path.stem}` into an implementation-ready OpenSpec change.831- Preserve traceability to the source idea while proposal, specs, design, and tasks are refined.832833## Source Idea834835- Path: `{rel_idea}`836- Promoted: {created}837838### Idea Content839840```md841{idea_content.rstrip()}842```843844## Capabilities845846### New Capabilities847- `{change}`: Initial capability placeholder created from the promoted idea. Refine before implementation.848849### Modified Capabilities850- None yet.851852## Impact853854- TBD: Fill in affected code, APIs, dependencies, or systems before implementation.855"""856 tasks = """## 1. Proposal Refinement857858- [ ] 1.1 Review the source idea and clarify scope.859- [ ] 1.2 Identify affected capabilities and write spec deltas.860- [ ] 1.3 Add design notes if implementation trade-offs are non-trivial.861862## 2. Implementation863864- [ ] 2.1 Implement the approved behavior.865- [ ] 2.2 Add or update tests.866- [ ] 2.3 Validate OpenSpec artifacts and test suite.867"""868 spec_dir = change_dir / "specs" / change869 spec_dir.mkdir(parents=True, exist_ok=True)870 spec = f"""# {title}871872## ADDED Requirements873874### Requirement: Promoted idea scope875The system SHALL implement the approved behavior described by the promoted idea after this placeholder is refined into concrete requirements.876877#### Scenario: Placeholder refinement878- **GIVEN** the idea `{idea_path.stem}` has been promoted into this change879- **WHEN** implementation begins880- **THEN** this placeholder requirement is replaced or refined with concrete, testable behavior before the change is completed881"""882 (change_dir / "proposal.md").write_text(proposal, encoding="utf-8")883 (change_dir / "tasks.md").write_text(tasks, encoding="utf-8")884 (spec_dir / "spec.md").write_text(spec, encoding="utf-8")885 metadata = {886 "schema": "spec-driven",887 "sourceIdea": rel_idea,888 "promotedAt": created,889 }890 (change_dir / ".openspec.yaml").write_text("schema: spec-driven\nsourceIdea: " + json.dumps(rel_idea) + "\npromotedAt: " + json.dumps(created) + "\n", encoding="utf-8")891892 return json.dumps({893 "ok": True,894 "idea": idea_path.stem,895 "change": change,896 "change_path": str(change_dir),897 "relative_change_path": _relative(change_dir, root),898 "workdir": str(root),899 "paths": {900 "proposal": str(change_dir / "proposal.md"),901 "tasks": str(change_dir / "tasks.md"),902 "spec": str(spec_dir / "spec.md"),903 "metadata": str(change_dir / ".openspec.yaml"),904 },905 "metadata": metadata,906 })907908909def openspec_change_create(args: dict, **kwargs) -> str:910 root, err, _context = _resolve_project(args)911 if err or root is None:912 return _error(err or "workdir could not be resolved")913 change = _slugify(str(args.get("change") or args.get("change_id") or ""), fallback="")914 title = str(args.get("title") or change.replace("-", " ").title()).strip()915 summary = str(args.get("summary") or f"Create change: {title}").strip()916 if not change or not _VALID_CHANGE_RE.match(change):917 return _error("change must be a kebab-case id")918 tasks = _coerce_string_list(args.get("tasks"))919 result, error = _write_change_scaffold(920 root,921 change,922 title,923 summary,924 tasks=tasks,925 with_spec=bool(args.get("with_spec")) or bool(tasks),926 force=bool(args.get("force")),927 )928 if error:929 return _error(error, change=change, change_path=str(_change_path(root, change)))930 return json.dumps(result)931932933def openspec_change_promote(args: dict, **kwargs) -> str:934 root, err, _context = _resolve_project(args)935 if err or root is None:936 return _error(err or "workdir could not be resolved")937 change = _slugify(str(args.get("change") or args.get("change_id") or ""), fallback="")938 if not change:939 return _error("change is required")940 change_dir, archived = _resolve_change_path(root, change)941 if change_dir is None or archived:942 return _error(f"active change not found: {change}")943 tasks_path = change_dir / "tasks.md"944 if not tasks_path.exists() or bool(args.get("replace_tasks")):945 task_items = _coerce_string_list(args.get("tasks")) or _default_tasks()946 tasks_path.write_text(_format_tasks(task_items), encoding="utf-8")947 title = change.replace("-", " ").title()948 proposal = _read_doc(change_dir / "proposal.md")949 for line in proposal.splitlines():950 if line.startswith("# "):951 title = line.lstrip("# ").strip() or title952 break953 spec_path = _write_placeholder_spec(change_dir, change, title)954 tasks = _parse_tasks(tasks_path)955 return json.dumps({956 "ok": True,957 "change": change,958 "change_path": str(change_dir),959 "relative_change_path": _relative(change_dir, root),960 "workdir": str(root),961 "status": _derive_status(tasks_path),962 "counts": _task_counts(tasks),963 "paths": {"tasks": str(tasks_path), "spec": str(spec_path)},964 })965966967def openspec_change_sequence_set(args: dict, **kwargs) -> str:968 """Set dashboard-local order/dependency metadata for active changes."""969 root, err, context = _resolve_project(args)970 if err or root is None:971 return _error(err or "workdir could not be resolved")972 registry = _registry_module()973 if registry is None or not hasattr(registry, "set_change_sequence"):974 return _error("OpenSpec registry sequence support unavailable")975 source = context.get("source") if isinstance(context, dict) else None976 if source is None:977 source = registry.get_source_by_path(str(root))978 if source is None:979 return _error("sequence metadata requires a registered OpenSpec source; pass identifier/project or register the repo in the dashboard", workdir=str(root))980 raw_changes = _coerce_string_list(args.get("changes"))981 changes, missing = _resolve_change_names(root, registry, raw_changes)982 if missing:983 return _error("some changes were not found", missing=missing, workdir=str(root))984 if not changes:985 return _error("changes is required")986 raw_deps = args.get("dependencies") or args.get("depends_on") or {}987 if raw_deps is None:988 raw_deps = {}989 if not isinstance(raw_deps, dict):990 return _error("dependencies must be an object mapping change names/tokens to arrays of prerequisite change names/tokens")991 dependencies: dict[str, list[str]] = {}992 dep_missing: list[str] = []993 for raw_key, raw_values in raw_deps.items():994 key_names, key_missing = _resolve_change_names(root, registry, [str(raw_key)])995 dep_names, missing_deps = _resolve_change_names(root, registry, _coerce_string_list(raw_values))996 if key_missing:997 dep_missing.extend(key_missing)998 continue999 if missing_deps:1000 dep_missing.extend(missing_deps)1001 if key_names:1002 dependencies[key_names[0]] = dep_names1003 if dep_missing:1004 return _error("some dependency changes were not found", missing=dep_missing, workdir=str(root))1005 source_id = str(source.get("id") or source.get("token"))1006 group_id = str(args.get("group_id") or args.get("groupId") or "default").strip() or "default"1007 sequence = registry.set_change_sequence(source_id, changes, group_id=group_id, dependencies=dependencies)1008 return json.dumps({1009 "ok": True,1010 "source": registry.effective_name(source),1011 "source_id": source_id,1012 "workdir": str(root),1013 "groupId": group_id,1014 "changes": [{"name": name, "token": registry.change_token(name), "sequence": sequence.get(name, {})} for name in changes],1015 "note": "Sequence/dependency metadata is stored in the Hermes OpenSpec plugin DB, not in OpenSpec files.",1016 })101710181019def openspec_task_list(args: dict, **kwargs) -> str:1020 root, err, _context = _resolve_project(args)1021 if err or root is None:1022 return _error(err or "workdir could not be resolved")1023 change = _slugify(str(args.get("change") or args.get("change_id") or ""), fallback="")1024 if not change:1025 return _error("change is required")1026 change_dir, archived = _resolve_change_path(root, change)1027 if change_dir is None:1028 return _error(f"change not found: {change}")1029 tasks_path = change_dir / "tasks.md"1030 tasks = _parse_tasks(tasks_path)1031 if not tasks:1032 return _error(f"tasks not found for change: {change}", change=change, tasks=[])1033 return json.dumps({1034 "ok": True,1035 "change": change,1036 "archived": archived,1037 "path": str(tasks_path),1038 "relative_path": _relative(tasks_path, root),1039 "tasks": tasks,1040 "counts": _task_counts(tasks),1041 "status": _derive_status(tasks_path, archived=archived),1042 })104310441045def openspec_task_set_status(args: dict, **kwargs) -> str:1046 root, err, _context = _resolve_project(args)1047 if err or root is None:1048 return _error(err or "workdir could not be resolved")1049 change = _slugify(str(args.get("change") or args.get("change_id") or ""), fallback="")1050 status = str(args.get("status") or "").strip().lower()1051 if status in {"open", "todo", "pending"}:1052 mark = " "1053 normalized = "todo"1054 elif status in {"done", "complete", "completed"}:1055 mark = "x"1056 normalized = "done"1057 else:1058 return _error("status must be 'todo' or 'done'")1059 task_ids = set(_coerce_string_list(args.get("tasks") or args.get("task_ids")))1060 if not task_ids:1061 return _error("tasks is required")1062 change_dir, archived = _resolve_change_path(root, change)1063 if change_dir is None or archived:1064 return _error(f"active change not found: {change}")1065 tasks_path = change_dir / "tasks.md"1066 if not tasks_path.is_file():1067 return _error(f"tasks not found for change: {change}")1068 lines = tasks_path.read_text(encoding="utf-8", errors="replace").splitlines()1069 # Build a lookup of parsed tasks so we can match by numeric ID or text substring.1070 parsed = _parse_tasks(tasks_path)1071 # Resolve each requested task_id to actual line numbers.1072 # Accept: numeric IDs (exact match), or text substrings (case-insensitive).1073 target_lines: set[int] = set()1074 not_found: list[str] = []1075 for requested in task_ids:1076 # Try exact numeric ID match first1077 id_matches = [t for t in parsed if t["id"] == requested]1078 if id_matches:1079 for t in id_matches:1080 target_lines.add(t["line"])1081 continue1082 # Try text substring match (case-insensitive)1083 needle = requested.lower()1084 text_matches = [t for t in parsed if needle in t["text"].lower()]1085 if text_matches:1086 for t in text_matches:1087 target_lines.add(t["line"])1088 continue1089 not_found.append(requested)1090 if not_found:1091 return _error("task ids not found: " + ", ".join(not_found), missing=not_found)1092 # Rewrite matching lines1093 updated_lines = []1094 matched_ids: set[str] = set()1095 for line_no, line in enumerate(lines, start=1):1096 if line_no in target_lines:1097 match = _TASK_LINE_RE.match(line)1098 if match:1099 matched_ids.add(match.group("id") or str(line_no))1100 line = f"{match.group('prefix')}{mark}{match.group('suffix')}"1101 updated_lines.append(line)1102 tasks_path.write_text("\n".join(updated_lines) + "\n", encoding="utf-8")1103 tasks = _parse_tasks(tasks_path)1104 return json.dumps({1105 "ok": True,1106 "change": change,1107 "updated": sorted(matched_ids),1108 "set_status": normalized,1109 "path": str(tasks_path),1110 "relative_path": _relative(tasks_path, root),1111 "tasks": tasks,1112 "counts": _task_counts(tasks),1113 "status": _derive_status(tasks_path),1114 })111511161117def openspec_change_archive(args: dict, **kwargs) -> str:1118 root, err, _context = _resolve_project(args)1119 if err or root is None:1120 return _error(err or "workdir could not be resolved")1121 change = _slugify(str(args.get("change") or args.get("change_id") or ""), fallback="")1122 if not change:1123 return _error("change is required")1124 change_dir = _change_path(root, change)1125 if not change_dir.is_dir():1126 return _error(f"active change not found: {change}")1127 tasks_path = change_dir / "tasks.md"1128 status = _derive_status(tasks_path)1129 if status != "done" and not args.get("force"):1130 return _error(f"change is not complete: {change}", status=status)1131 archive_dir = _change_path(root, change, archived=True)1132 archive_dir.parent.mkdir(parents=True, exist_ok=True)1133 if archive_dir.exists() and not args.get("force"):1134 return _error(f"archived change already exists: {change}")1135 if archive_dir.exists():1136 shutil.rmtree(archive_dir)1137 shutil.move(str(change_dir), str(archive_dir))1138 return json.dumps({1139 "ok": True,1140 "change": change,1141 "status": "archived",1142 "archived_path": str(archive_dir),1143 "relative_archived_path": _relative(archive_dir, root),1144 "workdir": str(root),1145 })114611471148def openspec_change_unarchive(args: dict, **kwargs) -> str:1149 root, err, _context = _resolve_project(args)1150 if err or root is None:1151 return _error(err or "workdir could not be resolved")1152 change = _slugify(str(args.get("change") or args.get("change_id") or ""), fallback="")1153 if not change:1154 return _error("change is required")1155 archive_dir = _change_path(root, change, archived=True)1156 active_dir = _change_path(root, change)1157 if not archive_dir.is_dir():1158 return _error(f"archived change not found: {change}")1159 if active_dir.exists() and not args.get("force"):1160 return _error(f"active change already exists: {change}")1161 if active_dir.exists():1162 shutil.rmtree(active_dir)1163 shutil.move(str(archive_dir), str(active_dir))1164 status = _derive_status(active_dir / "tasks.md")1165 return json.dumps({1166 "ok": True,1167 "change": change,1168 "status": status,1169 "change_path": str(active_dir),1170 "relative_change_path": _relative(active_dir, root),1171 "workdir": str(root),1172 })117311741175def openspec_list(args: dict, **kwargs) -> str:1176 cmd = ["list", "--json"]1177 kind = str(args.get("kind") or "changes").strip().lower()1178 if kind == "specs":1179 cmd.append("--specs")1180 else:1181 cmd.append("--changes")1182 sort = str(args.get("sort") or "").strip().lower()1183 if sort in {"recent", "name"}:1184 cmd.extend(["--sort", sort])1185 return _run(cmd, args.get("workdir"))118611871188def openspec_show(args: dict, **kwargs) -> str:1189 name = str(args.get("name") or "").strip()1190 if not name:1191 return json.dumps({"ok": False, "error": "name is required"})1192 cmd = ["show", name, "--json", "--no-interactive"]1193 item_type = str(args.get("type") or "").strip().lower()1194 if item_type in {"change", "spec"}:1195 cmd.extend(["--type", item_type])1196 if args.get("deltas_only"):1197 cmd.append("--deltas-only")1198 if args.get("requirements_only"):1199 cmd.append("--requirements")1200 if args.get("no_scenarios"):1201 cmd.append("--no-scenarios")1202 requirement = str(args.get("requirement") or "").strip()1203 if requirement:1204 cmd.extend(["--requirement", requirement])1205 return _run(cmd, args.get("workdir"))120612071208def openspec_validate(args: dict, **kwargs) -> str:1209 cmd = ["validate", "--json", "--no-interactive"]1210 if args.get("strict", True):1211 cmd.append("--strict")1212 scope = str(args.get("scope") or "").strip().lower()1213 target = str(args.get("target") or "").strip()1214 if scope == "all" or (not target and not scope):1215 cmd.append("--all")1216 elif scope == "changes":1217 cmd.append("--changes")1218 elif scope == "specs":1219 cmd.append("--specs")1220 elif target:1221 cmd.append(target)1222 else:1223 return json.dumps({"ok": False, "error": "target is required when scope='target'"})1224 item_type = str(args.get("type") or "").strip().lower()1225 if item_type in {"change", "spec"}:1226 cmd.extend(["--type", item_type])1227 return _run(cmd, args.get("workdir"))122812291230def openspec_status(args: dict, **kwargs) -> str:1231 change = str(args.get("change") or "").strip()1232 if not change:1233 return json.dumps({"ok": False, "error": "change is required"})1234 cmd = ["status", "--change", change, "--json"]1235 schema = str(args.get("schema") or "").strip()1236 if schema:1237 cmd.extend(["--schema", schema])1238 return _run(cmd, args.get("workdir"))123912401241def openspec_instructions(args: dict, **kwargs) -> str:1242 cmd = ["instructions", "--json"]1243 artifact = _normalize_artifact(str(args.get("artifact") or ""))1244 if artifact:1245 cmd.append(artifact)1246 change = str(args.get("change") or "").strip()1247 if change:1248 cmd.extend(["--change", change])1249 schema = str(args.get("schema") or "").strip()1250 if schema:1251 cmd.extend(["--schema", schema])12521253 result = _run(cmd, args.get("workdir"))1254 parsed = json.loads(result)1255 if parsed.get("ok"):1256 return result1257 reason = "\n".join(str(parsed.get(key) or "") for key in ("stderr", "stdout", "error"))1258 fallback = _template_instruction_fallback(artifact, reason, args.get("workdir"), schema)1259 return fallback or result126012611262def openspec_spec_diff(args: dict, **kwargs) -> str:1263 """Compare a change's delta spec against its baseline, or a worktree spec1264 against its HEAD version, and return the structured semantic delta plus a1265 unified line diff fallback.12661267 Filesystem-backed — does not require the OpenSpec CLI binary.1268 """1269 root, err, _context = _resolve_project(args)1270 if err or root is None:1271 return _error(err or "workdir could not be resolved")12721273 spec = str(args.get("spec") or "").strip()1274 if not spec:1275 return _error("spec is required (e.g. 'agent-tools')")12761277 # Prevent path traversal1278 if spec.startswith("/") or ".." in Path(spec).parts:1279 return _error("spec must be a relative path without parent references")12801281 change = str(args.get("change") or "").strip()12821283 try:1284 from . import spec_parser # type: ignore1285 except ImportError:1286 import spec_parser # type: ignore12871288 if change:1289 # Diff change spec vs baseline1290 change_spec_path = root / "openspec" / "changes" / change / "specs" / f"{spec}" / "spec.md"1291 # Also handle nested paths like "agent-tools/spec.md"1292 if not change_spec_path.is_file():1293 change_spec_path = root / "openspec" / "changes" / change / "specs" / f"{spec}.md"1294 if not change_spec_path.is_file():1295 return _error(1296 f"change spec not found: openspec/changes/{change}/specs/{spec}/spec.md",1297 change=change,1298 spec=spec,1299 )13001301 after_md = _read_doc(change_spec_path)1302 baseline_path = root / "openspec" / "specs" / spec / "spec.md"1303 if not baseline_path.is_file():1304 baseline_path = root / "openspec" / "specs" / f"{spec}.md"13051306 before_md = _read_doc(baseline_path) if baseline_path.is_file() else None1307 baseline_exists = before_md is not None1308 else:1309 # Diff worktree spec vs HEAD1310 worktree_path = root / "openspec" / "specs" / spec / "spec.md"1311 if not worktree_path.is_file():1312 worktree_path = root / "openspec" / "specs" / f"{spec}.md"1313 if not worktree_path.is_file():1314 return _error(1315 f"spec not found: openspec/specs/{spec}/spec.md",1316 spec=spec,1317 )13181319 after_md = _read_doc(worktree_path)1320 before_md = _git_show_spec(root, spec)1321 if before_md is None:1322 return _error(1323 "git is required for worktree-vs-HEAD comparison and the spec "1324 "has no git history (it may be untracked). Pass 'change' to "1325 "diff against a change's delta spec instead.",1326 spec=spec,1327 )1328 baseline_exists = True13291330 diff = spec_parser.semantic_spec_diff(before_md, after_md)1331 line_diff = spec_parser.unified_diff(before_md, after_md, spec)13321333 return json.dumps({1334 "ok": True,1335 "spec": spec,1336 "change": change or None,1337 "status": diff["status"],1338 "baseline_exists": baseline_exists,1339 "requirements": diff["requirements"],1340 "line_diff": line_diff,1341 "workdir": str(root),1342 })134313441345def _git_show_spec(root: Path, spec_path: str) -> str | None:1346 """Return the HEAD version of a spec via git, or None if unavailable."""1347 try:1348 proc = subprocess.run(1349 ["git", "-C", str(root), "show", f"HEAD:openspec/specs/{spec_path}/spec.md"],1350 capture_output=True,1351 text=True,1352 timeout=8,1353 check=False,1354 )1355 except Exception:1356 return None1357 if proc.returncode != 0:1358 # Try flat path1359 try:1360 proc = subprocess.run(1361 ["git", "-C", str(root), "show", f"HEAD:openspec/specs/{spec_path}.md"],1362 capture_output=True,1363 text=True,1364 timeout=8,1365 check=False,1366 )1367 except Exception:1368 return None1369 if proc.returncode != 0:1370 return None1371 return proc.stdout137213731374def _spec_target_path(root: Path, spec: str, change: str) -> Path:1375 """Resolve the spec.md path for a given spec name and optional change scope."""1376 if change:1377 base = root / "openspec" / "changes" / change / "specs"1378 else:1379 base = root / "openspec" / "specs"1380 return base / spec / "spec.md"138113821383def _spec_dir_from_path(spec_path: Path) -> str:1384 """Extract the spec directory name from a spec.md path."""1385 return spec_path.parent.name138613871388def _list_specs_in_dir(specs_root: Path) -> list[str]:1389 """List spec names (directory names containing spec.md) under a specs root."""1390 if not specs_root.is_dir():1391 return []1392 names: list[str] = []1393 for child in sorted(specs_root.iterdir()):1394 if child.is_dir() and (child / "spec.md").is_file():1395 names.append(child.name)1396 elif child.is_file() and child.suffix == ".md" and child.name != "spec.md":1397 # Flat .md files at the root level1398 names.append(child.stem)1399 return names140014011402def openspec_spec_create(args: dict, **kwargs) -> str:1403 """Create a spec from structured input and write a formatted spec.md.14041405 Filesystem-backed — does not require the OpenSpec CLI binary.1406 """1407 root, err, _context = _resolve_project(args)1408 if err or root is None:1409 return _error(err or "workdir could not be resolved")14101411 spec = str(args.get("spec") or "").strip()1412 if not spec:1413 return _error("spec is required (e.g. 'agent-tools')")14141415 # Validate slug: kebab-case, no path traversal1416 if not _VALID_CHANGE_RE.match(spec) or ".." in Path(spec).parts or spec.startswith("/"):1417 return _error(f"spec must be kebab-case (letters, numbers, hyphens): {spec}")14181419 title = str(args.get("title") or "").strip()1420 if not title:1421 return _error("title is required")14221423 purpose = str(args.get("purpose") or "").strip()1424 if not purpose:1425 return _error("purpose is required")14261427 requirements = args.get("requirements")1428 if not isinstance(requirements, list) or not requirements:1429 return _error("requirements must be a non-empty array")14301431 change = str(args.get("change") or "").strip()1432 if change and not _VALID_CHANGE_RE.match(change):1433 return _error(f"change must be kebab-case: {change}")14341435 force = bool(args.get("force"))14361437 spec_path = _spec_target_path(root, spec, change)14381439 if spec_path.exists() and not force:1440 rel = _relative(spec_path, root)1441 return _error(1442 f"spec already exists at {rel}. Pass force=true to overwrite.",1443 spec=spec,1444 change=change or None,1445 path=rel,1446 )14471448 try:1449 from . import spec_parser # type: ignore1450 except ImportError:1451 import spec_parser # type: ignore14521453 md = spec_parser.spec_to_markdown(title, purpose, requirements)14541455 spec_path.parent.mkdir(parents=True, exist_ok=True)1456 spec_path.write_text(md, encoding="utf-8")14571458 return json.dumps({1459 "ok": True,1460 "spec": spec,1461 "change": change or None,1462 "path": _relative(spec_path, root),1463 "absolute_path": str(spec_path),1464 "workdir": str(root),1465 })146614671468def openspec_spec_show(args: dict, **kwargs) -> str:1469 """Read a spec as structured JSON using parse_spec.14701471 Filesystem-backed — does not require the OpenSpec CLI binary.1472 """1473 root, err, _context = _resolve_project(args)1474 if err or root is None:1475 return _error(err or "workdir could not be resolved")14761477 spec = str(args.get("spec") or "").strip()1478 if not spec:1479 return _error("spec is required (e.g. 'agent-tools')")14801481 if spec.startswith("/") or ".." in Path(spec).parts:1482 return _error("spec must be a relative path without parent references")14831484 change = str(args.get("change") or "").strip()14851486 spec_path = _spec_target_path(root, spec, change)1487 if not spec_path.is_file():1488 # Try flat .md path1489 if change:1490 flat = root / "openspec" / "changes" / change / "specs" / f"{spec}.md"1491 else:1492 flat = root / "openspec" / "specs" / f"{spec}.md"1493 if flat.is_file():1494 spec_path = flat1495 else:1496 location = f"openspec/changes/{change}/specs/{spec}/spec.md" if change else f"openspec/specs/{spec}/spec.md"1497 return _error(f"spec not found: {location}", spec=spec, change=change or None)14981499 try:1500 from . import spec_parser # type: ignore1501 except ImportError:1502 import spec_parser # type: ignore15031504 md = _read_doc(spec_path)1505 parsed = spec_parser.parse_spec(md)15061507 return json.dumps({1508 "ok": True,1509 "spec": spec,1510 "change": change or None,1511 "path": _relative(spec_path, root),1512 "title": parsed["title"],1513 "purpose": parsed["purpose"],1514 "requirements": parsed["requirements"],1515 "workdir": str(root),1516 })151715181519def openspec_spec_list(args: dict, **kwargs) -> str:1520 """List specs within a change, or baseline specs.15211522 Filesystem-backed — does not require the OpenSpec CLI binary.1523 """1524 root, err, _context = _resolve_project(args)1525 if err or root is None:1526 return _error(err or "workdir could not be resolved")15271528 change = str(args.get("change") or "").strip()15291530 if change:1531 if not _VALID_CHANGE_RE.match(change):1532 return _error(f"change must be kebab-case: {change}")1533 specs_root = root / "openspec" / "changes" / change / "specs"1534 else:1535 specs_root = root / "openspec" / "specs"15361537 names = _list_specs_in_dir(specs_root)15381539 return json.dumps({1540 "ok": True,1541 "change": change or None,1542 "specs": names,1543 "count": len(names),1544 "workdir": str(root),1545 })154615471548def openspec_cli(args: dict, **kwargs) -> str:1549 """Run the openspec CLI binary directly and return raw output.15501551 This is a passthrough tool — it runs ``openspec <command> [--json]`` and1552 returns the raw stdout. Use this when you need the CLI's native JSON format1553 (e.g. ``applyRequires``, ``artifacts``, ``contextFiles``) rather than the1554 plugin's wrapped JSON shapes.15551556 Gated by ``check_fn`` — only appears when the openspec binary is available.1557 """1558 import shlex15591560 exe = _openspec_bin()1561 if exe is None:1562 return _error("openspec CLI binary not found. Set OPENSPEC_BIN or install via npm.")15631564 raw_command = str(args.get("command") or "").strip()1565 if not raw_command:1566 return _error("command is required (e.g. 'status --change my-change')")15671568 try:1569 cmd_parts = shlex.split(raw_command)1570 except ValueError:1571 cmd_parts = raw_command.split()15721573 json_output = args.get("json_output")1574 if json_output is None:1575 json_output = True1576 if json_output:1577 cmd_parts = [*cmd_parts, "--json"]15781579 workdir = str(args.get("workdir") or "").strip()1580 if workdir:1581 root, err = _resolve_workdir(workdir)1582 if err or root is None:1583 return _error(err or "workdir could not be resolved")1584 cwd = str(root)1585 else:1586 cwd = None15871588 try:1589 proc = subprocess.run(1590 [exe, *cmd_parts],1591 capture_output=True,1592 text=True,1593 timeout=120,1594 check=False,1595 cwd=cwd,1596 )1597 except subprocess.TimeoutExpired:1598 return _error("openspec CLI command timed out (120s)")1599 except OSError as exc:1600 return _error(f"failed to run openspec CLI: {exc}")16011602 stdout = proc.stdout or ""1603 stderr = proc.stderr or ""16041605 if json_output and proc.returncode == 0:1606 try:1607 parsed = json.loads(stdout)1608 return json.dumps({1609 "ok": True,1610 "exit_code": proc.returncode,1611 "stdout": parsed,1612 "workdir": cwd,1613 })1614 except json.JSONDecodeError:1615 pass # Fall through to raw output16161617 return json.dumps({1618 "ok": proc.returncode == 0,1619 "exit_code": proc.returncode,1620 "stdout": stdout,1621 "stderr": stderr.strip() if stderr else "",1622 "workdir": cwd,1623 })