orchestrata_ultimate.verification.traceability

Reference › orchestrata_ultimate › verification › traceability

The @traces traceability engine: it finds @traces(spec, story, reqs) annotations in source code and builds a requirements traceability matrix (RTM). It is imported by the package root and by the compliance volk's _pipeline.

scan_traces(root_dir) walks every .py file under a directory (skipping __pycache__ and files that fail to parse), parses each with ast, and extracts decorator data with _extract_traces_from_decorator, _scan_class and _scan_function into TraceInfo records (spec, story, requirement ids, module, class, method, file, line). It reads the source as text and never imports it. validate_traces(trace_list, spec_reqs) cross-checks the annotations against a spec's requirement ids and returns a ValidationReport with verified, missing, orphan and duplicates, plus coverage_percent. generate_rtm combines traces with Requirement records into an RTMReport, and save_rtm writes rtm.json and rtm.md to an output directory. UserStory models a story with its acceptance criteria.

At a glance

Class diagram

Dependency diagram

API

Classes

TraceInfo (dataclass)

Decorators: dataclass · Line 46

Collected trace annotation from source code.

Requirement (dataclass)

Decorators: dataclass · Line 77

A single requirement from a spec.

UserStory (dataclass)

Decorators: dataclass · Line 89

A user story with acceptance criteria.

ValidationReport (dataclass)

Decorators: dataclass · Line 128

Result of cross-checking traces against spec requirements.

RTMReport (dataclass)

Decorators: dataclass · Line 178

Complete RTM report with entries and validation.

Functions

Source

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

"""Standalone traceability engine for spec verification.

Provides the @traces decorator, AST scanner, RTM generator, and validator.
Zero external orchestrata_ultimate imports -- this module is self-contained for
traceability annotation and verification.

Usage in production code::


    @traces(spec="001", story="US-001", reqs=["REQ-001-US-001-001"])
    class MyClass:
        ...

Migrated from spike/traceability.py to production package.
"""

import ast
import json
import re
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

# Deliberate RE-EXPORT, not an unused import: verification/__init__.py and
# multiple consumers import ``traces`` from THIS module (the verification-layer
# facade over the layer-0 canonical decorator). Removing it as "unused" broke
# the whole package's import chain on 2026-07-08 (112 collection errors).
from orchestrata_ultimate.core.base.traces import _REQ_ID_PATTERN, traces  # noqa: F401

# _REQ_ID_PATTERN is IMPORTED, not re-declared. This module used to carry its
# own copy, and that copy silently fell behind: it lacked the
# ``[A-Z]{2,4}-\d{3}-\d{3}`` alternative, so it rejected every canonical
# story-scoped id (``REQ-001-US-001-001``) while the two copies in
# ``core/base/traces.py`` and ``core/traceability_utils.py`` accepted them.
# Three copies of one grammar is three chances to diverge; this is now two.
# ``core/traceability_utils.py`` still holds the remaining duplicate.
#
# The US pattern stays local because it is deliberately LOOSER than the core
# one: this module accepts a bare story id (``US-001``) as well as the
# story-scoped form, which the core validator does not.
_US_ID_PATTERN = re.compile(r"^US-\d{3}(?:-\d{3})?$")


@dataclass
class TraceInfo:
    """Collected trace annotation from source code."""

    spec: str = ""
    story: str = ""
    reqs: list[str] = field(default_factory=list)
    module: str = ""
    class_name: str = ""
    method: str = ""
    file_path: str = ""
    line_number: int = 0

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary representation.

        Returns:
            Dict with all trace fields.
        """
        return {
            "spec": self.spec,
            "story": self.story,
            "reqs": self.reqs,
            "module": self.module,
            "class_name": self.class_name,
            "method": self.method,
            "file_path": self.file_path,
            "line_number": self.line_number,
        }


@dataclass
class Requirement:
    """A single requirement from a spec."""

    id: str
    description: str
    story_id: str = ""
    spec_id: str = ""
    priority: str = "NORMAL"
    category: str = ""


@dataclass
class UserStory:
    """A user story with acceptance criteria."""

    id: str
    title: str
    as_a: str = ""
    i_want: str = ""
    so_that: str = ""
    acceptance_criteria: list[str] = field(default_factory=list)
    requirements: list[Requirement] = field(default_factory=list)
    spec_id: str = ""

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary representation.

        Returns:
            Dict with all user story fields.
        """
        return {
            "id": self.id,
            "title": self.title,
            "as_a": self.as_a,
            "i_want": self.i_want,
            "so_that": self.so_that,
            "acceptance_criteria": self.acceptance_criteria,
            "requirements": [
                {
                    "id": r.id,
                    "description": r.description,
                    "priority": r.priority,
                    "category": r.category,
                }
                for r in self.requirements
            ],
            "spec_id": self.spec_id,
        }


@dataclass
class ValidationReport:
    """Result of cross-checking traces against spec requirements."""

    verified: list[str] = field(default_factory=list)
    missing: list[str] = field(default_factory=list)
    orphan: list[str] = field(default_factory=list)
    duplicates: dict[str, list[str]] = field(default_factory=dict)

    @property
    def total_verified(self) -> int:
        """Count of verified requirements."""
        return len(self.verified)

    @property
    def total_missing(self) -> int:
        """Count of missing requirements."""
        return len(self.missing)

    @property
    def total_orphan(self) -> int:
        """Count of orphan traces."""
        return len(self.orphan)

    @property
    def coverage_percent(self) -> float:
        """Coverage percentage of verified vs total requirements."""
        total = len(self.verified) + len(self.missing)
        if total == 0:
            return 100.0
        return round(len(self.verified) / total * 100, 1)

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary representation.

        Returns:
            Dict with all validation fields and computed metrics.
        """
        return {
            "verified": self.verified,
            "missing": self.missing,
            "orphan": self.orphan,
            "duplicates": self.duplicates,
            "total_verified": self.total_verified,
            "total_missing": self.total_missing,
            "total_orphan": self.total_orphan,
            "coverage_percent": self.coverage_percent,
        }


@dataclass
class RTMReport:
    """Complete RTM report with entries and validation."""

    entries: list[dict[str, Any]] = field(default_factory=list)
    validation: ValidationReport | None = None
    generated_at: str = ""
    project: str = ""

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary representation.

        Returns:
            Dict with entries, validation, and metadata.
        """
        return {
            "entries": self.entries,
            "validation": (self.validation.to_dict() if self.validation else None),
            "generated_at": self.generated_at,
            "project": self.project,
        }


def _validate_req_id(req_id: str) -> bool:
    """Check if a requirement ID matches the expected format.

    Args:
        req_id: Requirement ID string to validate.

    Returns:
        True if the ID matches REQ-NNN-NNN or US-NNN format.
    """
    return bool(_REQ_ID_PATTERN.match(req_id) or _US_ID_PATTERN.match(req_id))


def _extract_string_value(node: ast.expr) -> str:
    """Extract string value from an AST constant node.

    Args:
        node: AST expression node.

    Returns:
        String value if node is a string constant, else empty string.
    """
    if isinstance(node, ast.Constant) and isinstance(node.value, str):
        return node.value
    return ""


def _extract_list_values(node: ast.expr) -> list[str]:
    """Extract list of string values from an AST List node.

    Args:
        node: AST expression node.

    Returns:
        List of string values from the list elements.
    """
    if isinstance(node, ast.List):
        return [
            _extract_string_value(elt)
            for elt in node.elts
            if isinstance(elt, ast.Constant) and isinstance(elt.value, str)
        ]
    return []


def _extract_traces_from_decorator(
    call_node: ast.Call,
) -> dict[str, Any] | None:
    """Extract trace data from an AST decorator call node.

    Args:
        call_node: AST Call node representing a decorator.

    Returns:
        Dict with spec/story/reqs if this is a @traces decorator,
        None otherwise.
    """
    func = call_node.func
    if isinstance(func, ast.Name) and func.id == "traces":
        pass
    elif isinstance(func, ast.Attribute) and func.attr == "traces":
        pass
    else:
        return None
    result: dict[str, Any] = {"spec": "", "story": "", "reqs": []}
    for kw in call_node.keywords:
        if kw.arg == "spec":
            result["spec"] = _extract_string_value(kw.value)
        elif kw.arg == "story":
            result["story"] = _extract_string_value(kw.value)
        elif kw.arg == "reqs":
            result["reqs"] = _extract_list_values(kw.value)
    return result if result["reqs"] else None


def _file_path_to_module(file_path: Path, root_dir: Path) -> str:
    """Convert a file path to a Python module path.

    Args:
        file_path: Path to the Python file.
        root_dir: Root directory for relative path computation.

    Returns:
        Dotted module path string.
    """
    try:
        relative = file_path.relative_to(root_dir)
    except ValueError:
        relative = file_path
    parts = list(relative.with_suffix("").parts)
    return ".".join(parts)


def _is_method(
    func_node: ast.FunctionDef | ast.AsyncFunctionDef,
    tree: ast.AST,
) -> bool:
    """Check if a function node is a method inside a class.

    Args:
        func_node: The function definition AST node.
        tree: The full AST tree to search.

    Returns:
        True if func_node is directly inside a ClassDef body.
    """
    for node in ast.walk(tree):
        if isinstance(node, ast.ClassDef):
            for item in node.body:
                if item is func_node:
                    return True
    return False


def _scan_class(
    class_node: ast.ClassDef,
    module: str,
    file_path: str,
) -> list[TraceInfo]:
    """Scan a class definition for @traces decorators.

    Args:
        class_node: AST ClassDef node.
        module: Dotted module path.
        file_path: Absolute file path.

    Returns:
        List of TraceInfo for class-level and method-level traces.
    """
    results: list[TraceInfo] = []
    class_trace_data = None
    for dec in class_node.decorator_list:
        if isinstance(dec, ast.Call):
            data = _extract_traces_from_decorator(dec)
            if data:
                class_trace_data = data
                results.extend(
                    TraceInfo(
                        spec=data["spec"],
                        story=data["story"],
                        reqs=[req_id],
                        module=module,
                        class_name=class_node.name,
                        method="",
                        file_path=file_path,
                        line_number=class_node.lineno,
                    )
                    for req_id in data["reqs"]
                )
    for item in class_node.body:
        if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
            method_traces = _scan_function(
                item,
                module,
                class_node.name,
                file_path,
                class_trace_data,
            )
            results.extend(method_traces)
    return results


def _scan_function(
    func_node: ast.FunctionDef | ast.AsyncFunctionDef,
    module: str,
    class_name: str,
    file_path: str,
    parent_trace: dict[str, Any] | None = None,
) -> list[TraceInfo]:
    """Scan a function definition for @traces decorators.

    Args:
        func_node: AST FunctionDef or AsyncFunctionDef node.
        module: Dotted module path.
        class_name: Enclosing class name, or empty string.
        file_path: Absolute file path.
        parent_trace: Trace data from enclosing class, if any.

    Returns:
        List of TraceInfo for each traced requirement.
    """
    results: list[TraceInfo] = []
    for dec in func_node.decorator_list:
        if isinstance(dec, ast.Call):
            data = _extract_traces_from_decorator(dec)
            if data:
                spec = data["spec"] or (parent_trace["spec"] if parent_trace else "")
                story = data["story"] or (parent_trace["story"] if parent_trace else "")
                results.extend(
                    TraceInfo(
                        spec=spec,
                        story=story,
                        reqs=[req_id],
                        module=module,
                        class_name=class_name,
                        method=func_node.name,
                        file_path=file_path,
                        line_number=func_node.lineno,
                    )
                    for req_id in data["reqs"]
                )
    return results


def scan_traces(root_dir: str) -> list[TraceInfo]:
    """Walk source tree and collect all @traces annotations via AST.

    Args:
        root_dir: Root directory to scan.

    Returns:
        List of TraceInfo objects found in the source tree.
    """
    root = Path(root_dir)
    if not root.is_dir():
        return []
    results: list[TraceInfo] = []
    for py_file in sorted(root.rglob("*.py")):
        if "__pycache__" in str(py_file):
            continue
        try:
            source = py_file.read_text(encoding="utf-8")
            tree = ast.parse(source, filename=str(py_file))
        except SyntaxError, UnicodeDecodeError:
            continue
        module_name = _file_path_to_module(py_file, root.parent)
        for node in ast.walk(tree):
            if isinstance(node, ast.ClassDef):
                results.extend(_scan_class(node, module_name, str(py_file)))
            elif isinstance(
                node,
                (ast.FunctionDef, ast.AsyncFunctionDef),
            ):
                if not _is_method(node, tree):
                    results.extend(
                        _scan_function(
                            node,
                            module_name,
                            "",
                            str(py_file),
                        )
                    )
    return results


def generate_rtm(
    trace_records: list[TraceInfo],
    requirements: list[Requirement],
    project: str = "",
) -> RTMReport:
    """Generate RTM report from traces and requirements.

    Args:
        trace_records: From scan_traces().
        requirements: Full requirements list.
        project: Project name.

    Returns:
        RTMReport with entries, validation, metadata.
    """
    entries: list[dict[str, Any]] = []
    for trace in trace_records:
        for req_id in trace.reqs:
            location = f"{trace.file_path}:{trace.line_number}"
            entries.append(
                {
                    "req_id": req_id,
                    "spec": trace.spec,
                    "user_story": trace.story,
                    "module": trace.module,
                    "class_name": trace.class_name,
                    "method": trace.method,
                    "file_path": trace.file_path,
                    "line_number": trace.line_number,
                    "status": "VERIFIED",
                    "evidence": f"@traces in {location}",
                }
            )
    entries.sort(key=lambda e: e["req_id"])
    spec_req_ids = [r.id for r in requirements]
    validation = validate_traces(trace_records, spec_req_ids)
    return RTMReport(
        entries=entries,
        validation=validation,
        generated_at=datetime.now(UTC).isoformat(),
        project=project,
    )


def validate_traces(
    trace_list: list[TraceInfo],
    spec_reqs: list[str],
) -> ValidationReport:
    """Cross-check traces against the full requirement list.

    Args:
        trace_list: List of TraceInfo from scanning.
        spec_reqs: List of requirement IDs from the spec.

    Returns:
        ValidationReport with verified, missing, orphan, duplicates.
    """
    spec_req_set: set[str] = set(spec_reqs)
    annotated_reqs: set[str] = set()
    req_locations: dict[str, list[str]] = {}
    for trace in trace_list:
        for req_id in trace.reqs:
            annotated_reqs.add(req_id)
            location = (
                f"{trace.class_name}.{trace.method}"
                if trace.method
                else trace.class_name
            )
            if not location:
                location = trace.module
            req_locations.setdefault(req_id, []).append(location)
    verified = sorted(annotated_reqs & spec_req_set)
    missing = sorted(spec_req_set - annotated_reqs)
    orphan = sorted(annotated_reqs - spec_req_set)
    duplicates = {
        req_id: locations
        for req_id, locations in req_locations.items()
        if len(locations) > 1
    }
    return ValidationReport(
        verified=verified,
        missing=missing,
        orphan=orphan,
        duplicates=duplicates,
    )


def save_rtm(report: RTMReport, output_dir: str) -> dict[str, str]:
    """Write RTM to disk as rtm.md + rtm.json.

    Args:
        report: RTMReport to persist.
        output_dir: Directory to write files to.

    Returns:
        Dict with keys 'json' and 'markdown' pointing to paths.
    """
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    json_path = out / "rtm.json"
    json_path.write_text(
        json.dumps(report.to_dict(), indent=2) + "\n",
        encoding="utf-8",
    )
    md_content = _render_rtm_markdown(report)
    md_path = out / "rtm.md"
    md_path.write_text(md_content, encoding="utf-8")
    return {"json": str(json_path), "markdown": str(md_path)}


def _render_rtm_markdown(report: RTMReport) -> str:
    """Render an RTM report as a markdown string.

    Args:
        report: The RTMReport to render.

    Returns:
        Markdown-formatted string.
    """
    v = report.validation
    lines = [
        "# Requirements Traceability Matrix",
        "",
        f"**Project**: {report.project}",
        f"**Generated**: {report.generated_at}",
        "",
    ]
    if v:
        lines.extend(
            [
                "## Coverage Summary",
                "",
                "| Metric | Value |",
                "|--------|-------|",
                f"| Verified | {v.total_verified} |",
                f"| Missing | {v.total_missing} |",
                f"| Orphan | {v.total_orphan} |",
                f"| Coverage | {v.coverage_percent}% |",
                "",
            ]
        )
    if report.entries:
        lines.extend(
            [
                "## Traced Requirements",
                "",
                "| Req ID | Spec | Story | Class | Method | File |",
                "|--------|------|-------|-------|--------|------|",
            ]
        )
        for e in report.entries:
            method = e.get("method", "") or "-"
            cls = e.get("class_name", "") or "-"
            fp = e.get("file_path", "")
            if "src/" in fp:
                fp = "src/" + fp.split("src/", 1)[1]
            lines.append(
                f"| {e['req_id']} | {e['spec']} "
                f"| {e['user_story']} "
                f"| {cls} | {method} | {fp} |"
            )
        lines.append("")
    if v and v.missing:
        lines.extend(["## Missing Requirements", ""])
        for req_id in v.missing:
            lines.append(f"- {req_id}")
        lines.append("")
    if v and v.orphan:
        lines.extend(["## Orphan Traces", ""])
        for req_id in v.orphan:
            lines.append(f"- {req_id}")
        lines.append("")
    return "\n".join(lines)