"""Validator for requirements-discovery fan-out packets + index. Called by validators/validate-run.py when task_type == "requirements-discovery" and a `fan-out/` directory exists under the run dir. """ from __future__ import annotations import re import sys from dataclasses import dataclass, field from pathlib import Path _VALIDATORS_DIR = Path(__file__).resolve().parent for _ssot_dir in (_VALIDATORS_DIR.parent / "scripts", _VALIDATORS_DIR.parent / "python"): if _ssot_dir.is_dir() and str(_ssot_dir) not in sys.path: sys.path.insert(0, str(_ssot_dir)) from okstra_ctl.work_categories import WORK_CATEGORIES # noqa: E402 from okstra_ctl.fanout import topological_order, CycleError # noqa: E402 from okstra_ctl.scope_provenance import ( # noqa: E402 brief_citation_problem, brief_end_state_ids, brief_headings, parse_source, ) _NEXT_PHASES = ("error-analysis", "implementation-planning") _UNIT_RE = re.compile(r"unit-\d{3}") # index.md 번호목록 항목에서만 unit-NNN 을 추출 — 내러티브 문장 중복 방지 _INDEX_ENTRY_RE = re.compile(r"^\s*\d+\.\s+(unit-\d{3})\b", re.MULTILINE) # A `###` subsection inside the section still ends it: its prose is commentary, # not provenance, and parsing it as bullets rejects legitimate packets. _PROVENANCE_SECTION_RE = re.compile( r"^##\s+Requirement Provenance\s*$(?P.*?)(?=^#{2,}\s|\Z)", re.MULTILINE | re.DOTALL, ) _PROVENANCE_BULLET_RE = re.compile(r"^\s*[-*]\s+(?P.+?)\s*$", re.MULTILINE) def _check_provenance( pkt_name: str, text: str, errors: list[str], headings: set[str], end_state: set[str], ) -> None: """A fan-out unit must cite the brief line that demanded it. `derived:` is not admissible here: cross-packet derivation cannot be resolved from a single packet, so each unit anchors directly on the brief or on an okstra contract rule. `headings` empty means the brief was unavailable — the grammar still applies, only the heading-existence half degrades to unverifiable. `end_state` empty means the brief predates the end-state sections, so the older heading form is still the admissible one. """ section = _PROVENANCE_SECTION_RE.search(text) if not section: errors.append( f"{pkt_name}: missing `## Requirement Provenance` section — a unit must " "cite the brief line that demanded it" ) return bullets = [ b for b in _PROVENANCE_BULLET_RE.findall(section.group("body")) if b.strip() ] if not bullets: errors.append( f"{pkt_name}: `## Requirement Provenance` is empty — cite at least one " "`brief:EB-001` (or the legacy `brief:`) or `contract:`" ) return for raw in bullets: ref = parse_source(raw) if ref.kind not in ("brief", "contract"): errors.append( f"{pkt_name}: unrecognized provenance `{raw}` — a fan-out unit's " "source must be `brief:EB-001` (or the legacy `brief:`) " "or `contract:`" ) elif ref.kind == "brief": problem = brief_citation_problem(ref, headings, end_state) if problem: errors.append( f"{pkt_name}: {problem}. A fan-out unit becomes the brief for a " "whole downstream task — it must trace to a line the reporter wrote." ) @dataclass class ValidationResult: ok: bool errors: list[str] = field(default_factory=list) def _frontmatter(text: str) -> dict[str, str]: if not text.startswith("---"): return {} end = text.find("\n---", 3) if end == -1: return {} fm: dict[str, str] = {} last_key: str | None = None for line in text[3:end].splitlines(): if ":" in line and not line.startswith((" ", "\t")): k, _, v = line.partition(":") fm[k.strip()] = v.strip() last_key = k.strip() elif last_key and line.strip().startswith("- "): # YAML 블록리스트 항목 — 해당 키 값에 공백으로 이어 붙임 token = line.strip()[2:].strip() fm[last_key] = (fm.get(last_key, "") + " " + token).strip() return fm def _parse_deps(raw: str) -> list[str]: return _UNIT_RE.findall(raw or "") def validate_fanout(run_dir: Path, brief_path: Path | None = None) -> ValidationResult: run_dir = Path(run_dir) fo = run_dir / "fan-out" if not fo.is_dir(): return ValidationResult(ok=True) headings = brief_headings(brief_path) if brief_path is not None else set() end_state = brief_end_state_ids(brief_path) if brief_path is not None else set() errors: list[str] = [] units: dict[str, list[str]] = {} for pkt in sorted(fo.glob("unit-*.md")): text = pkt.read_text(encoding="utf-8") fm = _frontmatter(text) _check_provenance(pkt.name, text, errors, headings, end_state) uid = fm.get("unit-id", "") if uid != pkt.stem: errors.append(f"{pkt.name}: unit-id {uid!r} != filename stem {pkt.stem!r}") if fm.get("domain") not in WORK_CATEGORIES: errors.append(f"{pkt.name}: domain {fm.get('domain')!r} not in {WORK_CATEGORIES}") if fm.get("recommended-next-phase") not in _NEXT_PHASES: errors.append(f"{pkt.name}: recommended-next-phase not in {_NEXT_PHASES}") units[pkt.stem] = _parse_deps(fm.get("depends-on", "")) if not units: errors.append("fan-out/: directory exists but no unit-*.md packets found") return ValidationResult(ok=False, errors=errors) order: list[str] = [] try: order = topological_order(units) except CycleError as exc: errors.append(f"depends-on cycle: {exc}") except ValueError as exc: errors.append(f"depends-on unresolved: {exc}") index = fo / "index.md" if not index.is_file(): errors.append("fan-out/index.md missing") else: listed = _INDEX_ENTRY_RE.findall(index.read_text(encoding="utf-8")) if set(listed) != set(units): errors.append( f"index.md units {sorted(set(listed))} != packets {sorted(units)}" ) elif order and listed != order: errors.append(f"index.md order {listed} is not the topological order {order}") return ValidationResult(ok=not errors, errors=errors)