orchestrata_ultimate.semantic_graph.spec_trace_parser

Reference › orchestrata_ultimate › semantic_graph › spec_trace_parser

Parses Orchestrata specs and @traces annotations into plain data, ready for spec_trace_enricher to load into the shared semantic graph (memgraph). It reads files only and writes nothing. Three dataclasses carry the result, each with to_dict: RequirementInfo (requirement id, spec id, story id, title), StoryInfo and SpecInfo (spec id, slug, title, with its stories and requirements). parse_spec_file reads one spec.md, using _extract_metadata for the metadata block and _extract_stories for user stories, with story ids normalised to US-NNN form. parse_requirements_file reads a checklists/requirements.md, turning checkbox characters into a done, not done or unknown state. parse_all_specs walks a specs/ root and combines each spec with its checklist.

Code-side functions use collect_traces from core.traceability: collect_code_traces and collect_test_traces gather annotations from a source tree and a test tree. map_test_modules_to_source heuristically maps test_*.py files to the source modules they exercise, by candidate dotted names, and build_trace_index groups traces by requirement id.

At a glance

Class diagram

Dependency diagram

API

Classes

RequirementInfo (dataclass)

Decorators: dataclass · Line 22

A single requirement parsed from a spec checklist.

StoryInfo (dataclass)

Decorators: dataclass · Line 42

A user story parsed from a spec document.

SpecInfo (dataclass)

Decorators: dataclass · Line 66

A specification with its stories and requirements.

Functions

Source

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

"""Parse Orchestrata specs and @traces annotations into structured data.

This module extracts specs, user stories, and requirements from the
``specs/`` tree and pairs them with code traces collected from the source tree.
The output is used by :mod:`spec_trace_enricher` to populate the semantic graph.
"""

from __future__ import annotations

import logging
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.traceability import TraceInfo, collect_traces

logger = logging.getLogger(__name__)


@dataclass
class RequirementInfo:
    """A single requirement parsed from a spec checklist."""

    req_id: str
    spec_id: str
    story_id: str
    title: str
    checked: bool | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "req_id": self.req_id,
            "spec_id": self.spec_id,
            "story_id": self.story_id,
            "title": self.title,
            "checked": self.checked,
        }


@dataclass
class StoryInfo:
    """A user story parsed from a spec document."""

    story_id: str
    spec_id: str
    title: str
    role: str = ""
    want: str = ""
    so_that: str = ""
    acceptance_criteria: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        return {
            "story_id": self.story_id,
            "spec_id": self.spec_id,
            "title": self.title,
            "role": self.role,
            "want": self.want,
            "so_that": self.so_that,
            "acceptance_criteria": self.acceptance_criteria,
        }


@dataclass
class SpecInfo:
    """A specification with its stories and requirements."""

    spec_id: str
    slug: str
    title: str
    status: str = ""
    priority: str = ""
    author: str = ""
    version: str = ""
    components: str = ""
    stories: list[StoryInfo] = field(default_factory=list)
    requirements: list[RequirementInfo] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        return {
            "spec_id": self.spec_id,
            "slug": self.slug,
            "title": self.title,
            "status": self.status,
            "priority": self.priority,
            "author": self.author,
            "version": self.version,
            "components": self.components,
            "stories": [s.to_dict() for s in self.stories],
            "requirements": [r.to_dict() for r in self.requirements],
        }


# Patterns used to extract structured data from markdown.
_STORY_HEADING_RE = re.compile(
    r"^#{2,3}\s+(US-\d+):\s*(.*)$", re.IGNORECASE | re.MULTILINE
)
# The canonical story segment inside a requirement id (RUL-031).
_STORY_IN_REQ_ID_RE = re.compile(r"^REQ-\d+[a-z]?-(US-\d{3})-", re.IGNORECASE)
_REQ_RE = re.compile(
    r"^-\s+\[(?P<checked>[xX ]?)\]\s+\*\*(?P<req_id>REQ-\S+)\*\*:\s*(?P<title>.*)$",
    re.MULTILINE,
)
_METADATA_RE = re.compile(
    r"^-\s+\*\*(?P<key>[^*]+)\*\*:\s*(?P<value>.*)$", re.MULTILINE
)
_ROLE_RE = re.compile(r"\*\*As a\*\*\s*(.*)", re.IGNORECASE)
_WANT_RE = re.compile(r"\*\*I want(?: to)?\*\*\s*(.*)", re.IGNORECASE)
_SO_THAT_RE = re.compile(r"\*\*So that\*\*\s*(.*)", re.IGNORECASE)


def _parse_check_status(checkbox: str) -> bool | None:
    """Convert a markdown checkbox character to a boolean or None."""
    checked = checkbox.strip().lower()
    if checked == "x":
        return True
    if checked in ("", " "):
        return False
    return None


def _normalize_story_id(story_id: str) -> str:
    """Normalize a story identifier to canonical ``US-NNN`` form.

    The digits are zero-padded to three (RUL-031). Without the padding this
    returned whatever width the source happened to use, so a hand-written
    ``## US-1:`` heading produced ``US-1`` — an id that matches nothing in the
    migrated corpus and silently fails every downstream join.
    """
    story_id = story_id.strip().upper()
    match = re.search(r"US-?(\d+)", story_id)
    if match:
        return f"US-{match.group(1).zfill(3)}"
    return story_id


def parse_requirements_file(path: Path, spec_id: str = "") -> list[RequirementInfo]:
    """Parse a ``checklists/requirements.md`` file.

    Args:
        path: Path to the requirements markdown file.
        spec_id: Optional spec identifier. If omitted, it is inferred from the
            first requirement ID (``REQ-<spec_id>-...``).

    Returns:
        List of :class:`RequirementInfo` objects.
    """
    if not path.is_file():
        return []

    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError as exc:
        logger.warning("Cannot read requirements file %s: %s", path, exc)
        return []

    requirements: list[RequirementInfo] = []

    # Determine the current story from the nearest preceding heading.
    lines = text.splitlines()
    story_from_line: dict[int, str] = {}
    current = ""
    for idx, line in enumerate(lines):
        match = _STORY_HEADING_RE.match(line)
        if match:
            current = match.group(1).upper()
        story_from_line[idx] = current

    for idx, line in enumerate(lines):
        match = _REQ_RE.match(line)
        if not match:
            continue

        req_id = match.group("req_id").strip()
        title = match.group("title").strip()
        checked = _parse_check_status(match.group("checked"))
        inferred_spec = spec_id
        story = story_from_line.get(idx, "")

        if not inferred_spec and req_id.startswith("REQ-"):
            parts = req_id.split("-")
            if len(parts) >= 2:
                inferred_spec = parts[1]

        if not story and req_id.startswith("REQ-"):
            # Match the story SEGMENT, never split positionally. A canonical id
            # spells the story across two hyphen-separated parts
            # (``REQ-007-US-001-001`` -> parts[2] == "US", parts[3] == "001"),
            # so the old ``parts[2]`` read yielded the bare string "US" and
            # every requirement parsed out of the migrated corpus carried a
            # story id that joined to nothing.
            story_match = _STORY_IN_REQ_ID_RE.match(req_id)
            if story_match:
                story = story_match.group(1)

        requirements.append(
            RequirementInfo(
                req_id=req_id,
                spec_id=inferred_spec,
                story_id=_normalize_story_id(story),
                title=title,
                checked=checked,
            )
        )

    return requirements


def _extract_metadata(text: str) -> dict[str, str]:
    """Extract key/value pairs from a ``## Metadata`` block."""
    metadata: dict[str, str] = {}
    # Find the Metadata section and stop at the next level-2 heading.
    match = re.search(
        r"^##\s+Metadata\s*$(.*?)(?:^---|^##\s+)", text, re.MULTILINE | re.DOTALL
    )
    section = match.group(1) if match else ""
    for key, value in _METADATA_RE.findall(section):
        metadata[key.strip().lower()] = value.strip()
    return metadata


def _extract_stories(text: str, spec_id: str) -> list[StoryInfo]:
    """Extract user stories from a spec markdown body."""
    stories: list[StoryInfo] = []
    headings = list(_STORY_HEADING_RE.finditer(text))

    for idx, heading in enumerate(headings):
        story_id = _normalize_story_id(heading.group(1))
        title = heading.group(2).strip()
        start = heading.end()
        end = headings[idx + 1].start() if idx + 1 < len(headings) else len(text)
        body = text[start:end]

        role = ""
        want = ""
        so_that = ""
        criteria: list[str] = []

        for line in body.splitlines():
            role_match = _ROLE_RE.search(line)
            if role_match:
                role = role_match.group(1).strip()
                continue
            want_match = _WANT_RE.search(line)
            if want_match:
                want = want_match.group(1).strip()
                continue
            so_that_match = _SO_THAT_RE.search(line)
            if so_that_match:
                so_that = so_that_match.group(1).strip()
                continue
            # Acceptance criteria bullets under a story.
            if line.strip().startswith("-") and "REQ-" not in line:
                criteria.append(line.strip().lstrip("-").strip())

        stories.append(
            StoryInfo(
                story_id=story_id,
                spec_id=spec_id,
                title=title,
                role=role,
                want=want,
                so_that=so_that,
                acceptance_criteria=criteria,
            )
        )

    return stories


def parse_spec_file(path: Path) -> SpecInfo | None:
    """Parse a ``spec.md`` file into a :class:`SpecInfo`.

    Args:
        path: Path to the spec markdown file.

    Returns:
        A :class:`SpecInfo` or ``None`` if the file cannot be read.
    """
    if not path.is_file():
        return None

    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError as exc:
        logger.warning("Cannot read spec file %s: %s", path, exc)
        return None

    metadata = _extract_metadata(text)
    spec_id = metadata.get("spec id", "")
    title = metadata.get("title", path.parent.name)
    slug = path.parent.name

    if not spec_id:
        # Try to infer from the directory name ``XXX-name``.
        parts = slug.split("-", 1)
        if parts and parts[0].isdigit():
            spec_id = parts[0]

    stories = _extract_stories(text, spec_id)

    return SpecInfo(
        spec_id=spec_id,
        slug=slug,
        title=title,
        status=metadata.get("status", ""),
        priority=metadata.get("priority", ""),
        author=metadata.get("author", ""),
        version=metadata.get("version", ""),
        components=metadata.get("components", ""),
        stories=stories,
        requirements=[],
    )


def parse_all_specs(specs_root: Path) -> list[SpecInfo]:
    """Parse every ``spec.md`` and its ``checklists/requirements.md``.

    Args:
        specs_root: Root of the specs tree (e.g. ``Path("specs")``).

    Returns:
        List of :class:`SpecInfo` objects with requirements attached.
    """
    if not specs_root.is_dir():
        raise ValueError(f"Not a directory: {specs_root}")

    specs: list[SpecInfo] = []

    for spec_dir in sorted(specs_root.iterdir()):
        if not spec_dir.is_dir():
            continue

        spec_path = spec_dir / "spec.md"
        if not spec_path.is_file():
            continue

        spec = parse_spec_file(spec_path)
        if spec is None:
            continue

        reqs_path = spec_dir / "checklists" / "requirements.md"
        spec.requirements = parse_requirements_file(reqs_path, spec.spec_id)
        specs.append(spec)

    return specs


def collect_code_traces(source_root: Path) -> list[TraceInfo]:
    """Collect @traces annotations from a Python source tree.

    This is a thin wrapper around
    :func:`orchestrata_ultimate.core.traceability.collect_traces` that accepts a
    :class:`Path`.

    Args:
        source_root: Root directory to scan (e.g. ``Path("src")``).

    Returns:
        List of :class:`TraceInfo` objects.
    """
    return collect_traces(str(source_root))


def collect_test_traces(tests_root: Path) -> list[TraceInfo]:
    """Collect @traces annotations from a Python test tree.

    Identical scanning mechanism to :func:`collect_code_traces`, pointed at
    ``tests/`` instead of ``src/`` so traced test functions/classes surface as
    :class:`TraceInfo` too (``audit item 5`` -- TESTS edges).

    Args:
        tests_root: Root directory to scan (e.g. ``Path("tests")``).

    Returns:
        List of :class:`TraceInfo` objects found in the test tree.
    """
    return collect_traces(str(tests_root))


# Suffixes that generator/coverage-wave test files commonly append after the
# module name they exercise (e.g. ``test_pipeline_extended.py`` for
# ``pipeline.py``). Stripped, longest-first, when deriving a heuristic
# test-file -> source-module mapping.
_TEST_STEM_SUFFIXES = (
    "_generated",
    "_extended_coverage",
    "_extended",
    "_behavioral",
    "_expanded",
    "_coverage",
    "_regression",
    "_edge_cases",
    "_signatures",
    "_defect_regressions",
    "_real",
    "_extra",
    "_more",
    "_pure",
    "_full",
)


def _candidate_module_names(rel_parts: tuple[str, ...], stem: str) -> list[str]:
    """Return candidate dotted source-module names for a ``test_<mod>`` stem.

    ``stem`` is the test file's basename without ``.py``. Candidates are
    generated by stripping common test-suffixes (``_extended``,
    ``_generated``, ...) from the part after ``test_``, longest suffix
    stripped first, then rooted under ``orchestrata_ultimate.<rel_parts>``.
    """
    if not stem.startswith("test_"):
        return []
    base = stem[len("test_") :]
    if not base:
        return []

    candidates = [base]
    for suffix in _TEST_STEM_SUFFIXES:
        if base.endswith(suffix) and len(base) > len(suffix):
            candidates.append(base[: -len(suffix)])

    seen: list[str] = []
    for candidate in candidates:
        if candidate and candidate not in seen:
            seen.append(candidate)

    return [".".join(("orchestrata_ultimate", *rel_parts, name)) for name in seen]


def map_test_modules_to_source(
    tests_root: Path, known_modules: set[str]
) -> dict[str, list[Path]]:
    """Heuristically map ``test_*.py`` files to the source module they exercise.

    Convention: ``tests/<pkg...>/test_<mod>[_suffix].py`` maps to
    ``orchestrata_ultimate.<pkg...>.<mod>``. A mapping is only accepted when
    the computed module name is present in ``known_modules`` -- i.e. a module
    that already has at least one @traces-derived ``CodeElement`` -- which
    keeps the heuristic conservative and avoids inventing edges to modules
    that were never scanned.

    Args:
        tests_root: Root of the tests tree (e.g. ``Path("tests")``).
        known_modules: Dotted module names discovered from source @traces
            scanning (see :func:`collect_code_traces`).

    Returns:
        Mapping from source module dotted name to the list of test file
        paths (relative to ``tests_root``) heuristically associated with it.
    """
    if not tests_root.is_dir():
        return {}

    mapping: dict[str, list[Path]] = {}
    for test_file in sorted(tests_root.rglob("test_*.py")):
        if "__pycache__" in str(test_file):
            continue
        rel_path = test_file.relative_to(tests_root)
        rel_parts = tuple(rel_path.parent.parts)
        stem = test_file.stem

        for candidate in _candidate_module_names(rel_parts, stem):
            if candidate in known_modules:
                mapping.setdefault(candidate, []).append(rel_path)
                break

    return mapping


def build_trace_index(
    specs: list[SpecInfo], traces: list[TraceInfo]
) -> dict[str, list[TraceInfo]]:
    """Group code traces by requirement ID.

    Args:
        specs: Parsed specs.
        traces: Collected code traces.

    Returns:
        Mapping from requirement ID to the list of traces that reference it.
    """
    index: dict[str, list[TraceInfo]] = {}
    valid_reqs = {r.req_id for spec in specs for r in spec.requirements}

    for trace in traces:
        for req_id in trace.reqs:
            if req_id in valid_reqs:
                index.setdefault(req_id, []).append(trace)

    return index