orchestrata_ultimate.openspec.parse

Reference › orchestrata_ultimate › openspec › parse

parse is the line-based markdown parser of the OpenSpec support package (SPEC-104 FR-003 and FR-006). parse_spec turns a capability spec.md into an OpenSpecDoc, reading the purpose section (_extract_section) and the requirement blocks with their scenarios (_parse_requirements). parse_deltas turns a change spec.md into Delta blocks keyed by the operation in the ADDED, MODIFIED, REMOVED or RENAMED requirements headers. validate returns a list of violations for a document, an empty list meaning valid. The parser never raises on malformed input and returns whatever it could parse; validate likewise never raises. Types come from models; render is its inverse. The package and repo import it.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/openspec/parse.py

"""Parse OpenSpec markdown into models (SPEC-104 FR-003/006).

Line-based parser for capability specs (``### Requirement:`` + ``#### Scenario:``)
and change deltas (``## ADDED|MODIFIED|REMOVED|RENAMED Requirements``). Never
raises on malformed input — it returns whatever it could parse.
"""

from __future__ import annotations

import re

from orchestrata_ultimate.core.traceability import traces
from orchestrata_ultimate.openspec.models import (
    Delta,
    DeltaOp,
    OpenSpecDoc,
    Requirement,
    Scenario,
)

_RE_REQ = re.compile(r"^###\s+Requirement:\s*(.*)$")
_RE_SCEN = re.compile(r"^####\s+Scenario:\s*(.*)$")
_RE_H2 = re.compile(r"^##\s+(.*)$")
_RE_DELTA = re.compile(r"^##\s+(ADDED|MODIFIED|REMOVED|RENAMED)\s+Requirements\b")


@traces(spec="104", story="US-104-001", reqs=["REQ-104-004", "REQ-104-005"])
def _parse_requirements(lines: list[str]) -> list[Requirement]:
    """Parse ``### Requirement:`` blocks (with scenarios) from a line list."""
    reqs: list[Requirement] = []
    cur_req: Requirement | None = None
    cur_scen: Scenario | None = None
    for line in lines:
        mreq = _RE_REQ.match(line)
        if mreq:
            cur_req = Requirement(title=mreq.group(1).strip())
            reqs.append(cur_req)
            cur_scen = None
            continue
        if cur_req is None:
            continue
        mscen = _RE_SCEN.match(line)
        if mscen:
            cur_scen = Scenario(name=mscen.group(1).strip())
            cur_req.scenarios.append(cur_scen)
            continue
        if _RE_H2.match(line):  # a new H2 ends the current requirement block
            cur_req = None
            cur_scen = None
            continue
        if not line.strip():
            continue
        if cur_scen is not None:
            cur_scen.lines.append(line.rstrip())
        else:
            cur_req.text = (
                f"{cur_req.text} {line.strip()}".strip()
                if cur_req.text
                else line.strip()
            )
    return reqs


@traces(spec="104", story="US-104-001", reqs=["REQ-104-004"])
def _extract_section(lines: list[str], heading: str) -> str:
    """Return the text under ``## <heading>`` up to the next ``## `` heading."""
    out: list[str] = []
    capturing = False
    for line in lines:
        if _RE_H2.match(line):
            if capturing:
                break
            capturing = re.match(rf"^##\s+{re.escape(heading)}\s*$", line) is not None
            continue
        if capturing and line.strip():
            out.append(line.strip())
    return " ".join(out).strip()


@traces(spec="104", story="US-104-001", reqs=["REQ-104-004"])
def parse_spec(text: str, capability: str = "") -> OpenSpecDoc:
    """Parse an OpenSpec capability spec.md into an OpenSpecDoc."""
    lines = text.splitlines()
    return OpenSpecDoc(
        capability=capability,
        purpose=_extract_section(lines, "Purpose"),
        requirements=_parse_requirements(lines),
    )


@traces(spec="104", story="US-104-001", reqs=["REQ-104-005"])
def parse_deltas(text: str) -> list[Delta]:
    """Parse a change delta spec.md into Delta blocks keyed by operation."""
    sections: list[tuple[DeltaOp, list[str]]] = []
    cur: list[str] | None = None
    for line in text.splitlines():
        m = _RE_DELTA.match(line)
        if m:
            cur = []
            sections.append((m.group(1), cur))  # type: ignore[arg-type]
        elif cur is not None:
            cur.append(line)
    return [
        Delta(operation=op, requirements=_parse_requirements(ls)) for op, ls in sections
    ]


@traces(spec="104", story="US-104-001", reqs=["REQ-104-009"])
def validate(doc: OpenSpecDoc) -> list[str]:
    """Return OpenSpec validation violations (empty list = valid). Never raises."""
    violations: list[str] = []
    if not doc.requirements:
        violations.append("spec has no requirements")
    for req in doc.requirements:
        if not req.scenarios:
            violations.append(
                f"requirement {req.title!r} has no scenario (OpenSpec requires >=1)"
            )
        if "SHALL" not in req.text.upper():
            violations.append(f"requirement {req.title!r} has no SHALL statement")
    return violations