orchestrata_ultimate.core.traceability_utils

Reference › orchestrata_ultimate › core › traceability_utils

A standalone requirements-traceability engine. It has no Orchestrata imports, because the module is copied into generated project workspaces, where generated code does from traceability import traces. It provides four things: the traces(spec, story, reqs) decorator, an AST scanner, an RTM (requirements traceability matrix) generator and a validator.

The data model is a set of dataclasses: TraceInfo (one annotation with module, class, method, file and line), Requirement, UserStory, ValidationReport (verified, missing, orphan and duplicate requirement ids, with counts and a coverage percentage) and RTMReport. scan_traces(root_dir) walks a source tree and collects annotations from decorators by parsing, without importing the code (_scan_class, _scan_function and _extract_traces_from_decorator do the work, and a method inherits its class's annotation). validate_traces compares collected traces with a list of specification requirement ids; _validate_req_id checks an id against the accepted REQ- patterns. generate_rtm builds the report, and save_rtm writes rtm.md and rtm.json through atomic_json. The project's own traceability code is in core.traceability and core.tracing.

At a glance

Class diagram

Dependency diagram

API

Classes

TraceInfo (dataclass)

Decorators: dataclass · Line 31

Collected trace annotation from source code.

Requirement (dataclass)

Decorators: dataclass · Line 57

A single requirement from a spec.

UserStory (dataclass)

Decorators: dataclass · Line 69

A user story with acceptance criteria.

ValidationReport (dataclass)

Decorators: dataclass · Line 103

Result of cross-checking traces against spec requirements.

RTMReport (dataclass)

Decorators: dataclass · Line 144

Complete RTM report with entries and validation.

Functions

Source

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

# mypy: ignore-errors
"""Standalone traceability engine for spike-generated projects.

Provides the @traces decorator, AST scanner, RTM generator, and validator.
Zero orchestrata imports -- this module is copied into generated project
workspaces.

Usage in generated code:
    from traceability import traces

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

import ast
import re
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

_REQ_ID_PATTERN = re.compile(
    r"^REQ-\d{3}-(?:\d{3}(?:-\d{3})?|[A-Z]{2,4}\d*-\d{3}|[A-Z]{2,4}-\d{3}-\d{3}|C\d+|Exit-Criteria-for-Th-\d{3})$"
)
_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]:
        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]:
        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:
        return len(self.verified)

    @property
    def total_missing(self) -> int:
        return len(self.missing)

    @property
    def total_orphan(self) -> int:
        return len(self.orphan)

    @property
    def coverage_percent(self) -> float:
        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]:
        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]:
        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."""
    return bool(_REQ_ID_PATTERN.match(req_id) or _US_ID_PATTERN.match(req_id))


def traces(
    spec: str = "",
    story: str = "",
    reqs: list[str] | None = None,
) -> Callable:
    """Decorator that annotates classes/methods with traceability.

    Args:
        spec: Specification ID (e.g., "001").
        story: User story ID (e.g., "US-001").
        reqs: List of requirement IDs.

    Returns:
        The original class/function with __traces__ attached.
    """
    if reqs is None:
        reqs = []
    if not reqs:
        raise ValueError("@traces requires at least one requirement ID in reqs=[]")
    for req_id in reqs:
        if not _validate_req_id(req_id):
            raise ValueError(
                f"Invalid requirement ID: '{req_id}'. "
                f"Expected REQ-NNN-NNN[-NNN], REQ-NNN-ACn-NNN, REQ-NNN-FRn-NNN, "
                f"REQ-NNN-USn-NNN, REQ-NNN-NFRn-NNN, REQ-NNN-Cn (cluster/phase id, "
                f"e.g. REQ-033-C1), or US-NNN-NNN."
            )

    def decorator(obj: Any) -> Any:
        existing = getattr(obj, "__traces__", None)
        new_entry = {"spec": spec, "story": story, "reqs": list(reqs)}
        if existing is None:
            obj.__traces__ = new_entry
        elif isinstance(existing, list):
            existing.append(new_entry)
        else:
            obj.__traces__ = [existing, new_entry]
        return obj

    return decorator


@traces(spec="026", story="Step 1 — Trace Scanning", reqs=["REQ-026-006"])
def _extract_string_value(node: ast.expr) -> str:
    if isinstance(node, ast.Constant) and isinstance(node.value, str):
        return node.value
    return ""


@traces(spec="026", story="Step 1 — Trace Scanning", reqs=["REQ-026-006"])
def _extract_list_values(node: ast.expr) -> list[str]:
    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 []


@traces(spec="026", story="Step 1 — Trace Scanning", reqs=["REQ-026-006"])
def _extract_traces_from_decorator(
    call_node: ast.Call,
) -> dict[str, Any] | None:
    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


@traces(spec="026", story="Step 1 — Trace Scanning", reqs=["REQ-026-006"])
def _file_path_to_module(file_path: Path, root_dir: Path) -> str:
    try:
        relative = file_path.relative_to(root_dir)
    except ValueError:
        relative = file_path
    parts = list(relative.with_suffix("").parts)
    return ".".join(parts)


@traces(spec="026", story="Step 1 — Trace Scanning", reqs=["REQ-026-006"])
def _is_method(
    func_node: ast.FunctionDef | ast.AsyncFunctionDef, tree: ast.AST
) -> bool:
    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


@traces(spec="026", story="Step 1 — Trace Scanning", reqs=["REQ-026-006"])
def _scan_class(
    class_node: ast.ClassDef, module: str, file_path: str
) -> list[TraceInfo]:
    results = []
    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


@traces(spec="026", story="Step 1 — Trace Scanning", reqs=["REQ-026-006"])
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]:
    results = []
    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


@traces(spec="026", story="Step 1 — Trace Scanning", reqs=["REQ-026-006"])
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.
    """
    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


@traces(spec="026", story="Step 3 — RTM Generation", reqs=["REQ-026-014"])
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 = []
    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,
    )


@traces(spec="026", story="Step 3 — RTM Generation", reqs=["REQ-026-014"])
def validate_traces(
    trace_list: list[TraceInfo],
    spec_reqs: list[str],
) -> ValidationReport:
    """Cross-check traces against the full requirement list."""
    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,
    )


@traces(spec="026", story="Step 6 — RTM Persistence", reqs=["REQ-026-026"])
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'.
    """
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    # Atomic: both are registered artifacts regenerated by the gate battery
    # while the tree is live; a truncate-in-place rewrite keeps the inode and
    # SIGBUSes a reader holding it mmap'd. Byte-identical to the previous
    # writers (indent=2 plus the explicit trailing newline; markdown as-is).
    from orchestrata_ultimate.core.atomic_json import (
        dump_json_atomic,
        dump_text_atomic,
    )

    json_path = out / "rtm.json"
    dump_json_atomic(json_path, report.to_dict(), indent=2, trailing_newline=True)
    md_content = _render_rtm_markdown(report)
    md_path = out / "rtm.md"
    dump_text_atomic(md_path, md_content)
    return {"json": str(json_path), "markdown": str(md_path)}


@traces(spec="026", story="Step 6 — RTM Persistence", reqs=["REQ-026-026"])
def _render_rtm_markdown(report: RTMReport) -> str:
    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)