orchestrata_ultimate.agents.volks.compliance._pipeline

Reference › orchestrata_ultimate › agents › volks › compliance › _pipeline

_pipeline contains the standalone steps of the compliance run; volk is the only importer and calls them in order. The steps are module functions over a knowledge-graph store: scan_traces collects @traces records from the source and test directories; load_requirements reads requirements for the project from the graph (_fetch_kg_requirements converts them to dataclasses); generate_rtm combines traces and requirements into a traceability report; sync_code writes the code trace records to the graph; check_compliance checks design against code through the graph; and persist_rtm saves the report to disk as JSON and markdown (_save_rtm_to_disk) and as a graph snapshot (_save_rtm_to_kg). _task_slug and _project_slug produce the URL-safe names used to scope the run's data in the graph.

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/agents/volks/compliance/_pipeline.py

"""KG pipeline functions for ComplianceVolk -- standalone helpers.

Extracted from ComplianceVolk to keep the class under 300 lines.
All functions take dependencies explicitly; none mutate shared state.
"""

import logging
import os
import re
import uuid
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)

_SLUG_PATTERN = re.compile(r"[^\w-]")
_SLUG_MAX_LEN = 50


@traces(spec="026", story="Task Slug", reqs=["REQ-026-035"])
def _task_slug(task_summary: str) -> str:
    """Convert a task summary to a URL-safe slug.

    Cosmetic/display use ONLY (e.g. the RTM report's ``project`` label in
    ``generate_rtm``) -- never for KG project-scoping. A free-text task/
    intent summary is neither stable nor unique per project, so slugifying
    it was previously (mis)used as the KG ``project_slug`` too: an empty
    ``task_summary`` produced ``project_slug=""`` and a long one produced a
    silently-truncated-to-50-chars slug, each MERGEing its own anonymous
    ``:Project`` hub in the live graph that thousands of CodeClass/
    CodeFunction ``DEFINES`` edges then anchored onto (2026-07 audit). KG
    writes now resolve ``_project_slug(project_dir)`` instead — see below.

    Args:
        task_summary: Human-readable task description.

    Returns:
        Lowercase slug suitable for display.
    """
    raw = _SLUG_PATTERN.sub("-", task_summary.lower())
    return raw[:_SLUG_MAX_LEN].strip("-")


def _project_slug(project_dir: str) -> str:
    """Canonical KG project-scoping slug for this compliance run.

    Resolves via ``knowledge_graph.project_resolver.derive_project_id`` --
    the same identity scheme every other KG writer uses (project-directory
    basename, not free text) -- so this pipeline's Requirement lookups and
    CodeClass/CodeFunction sync land on the SAME Project hub the analyst
    volk (and every other KG writer) uses for this project. Never returns
    an empty string: falls back to the ``"unscoped"`` sentinel (matching
    ``kernel/pipeline_kg.py``'s established fallback) when ``project_dir``
    is unavailable, so ``KGDesignStore``'s empty-slug guard never fires from
    here.

    Args:
        project_dir: Project root directory (``ComplianceVolk.project_dir``,
            from the CXP packet's ``project_dir`` payload field).

    Returns:
        A non-empty project slug.
    """
    from orchestrata_ultimate.knowledge_graph.project_resolver import (
        derive_project_id,
    )

    return derive_project_id(project_dir) or "unscoped"


@traces(
    spec="026",
    story="Step 1 — Trace Scanning",
    reqs=["REQ-026-006", "REQ-026-007", "REQ-026-008", "REQ-026-009"],
)
def scan_traces(src_dir: str, tests_dir: str) -> list[Any]:
    """Scan @traces in source and test code.

    Args:
        src_dir: Path to source directory.
        tests_dir: Path to test directory.

    Returns:
        List of TraceInfo objects found.
    """
    try:
        from orchestrata_ultimate.verification.traceability import (
            scan_traces as _scan,
        )

        records: list[Any] = []
        if os.path.isdir(src_dir):
            records.extend(_scan(src_dir))
            logger.info("Source traces: %d found", len(records))
        if os.path.isdir(tests_dir):
            test_records = _scan(tests_dir)
            records.extend(test_records)
            logger.info("Test traces: %d found", len(test_records))
        return records
    except Exception as exc:
        logger.warning("Trace scan error: %s", exc)
        return []


@traces(
    spec="026",
    story="Step 2 — Requirements Loading",
    reqs=[
        "REQ-026-010",
        "REQ-026-011",
        "REQ-026-012",
        "REQ-026-013",
    ],
)
def load_requirements(kg_store: Any, project_dir: str) -> list[Any]:
    """Load requirements from knowledge graph.

    Args:
        kg_store: KGDesignStore instance, or None.
        project_dir: Project root directory for canonical KG lookup
            (``_project_slug`` -- NOT the task summary; see ``_task_slug``'s
            docstring for why free text was never a sound project key).

    Returns:
        List of Requirement objects, or empty list.
    """
    if not kg_store:
        return []
    try:
        return _fetch_kg_requirements(kg_store, project_dir)
    except Exception as exc:
        logger.warning("KG requirements load error: %s", exc)
        return []


@traces(
    spec="026",
    story="Step 2 — Requirements Loading",
    reqs=["REQ-026-010", "REQ-026-012"],
)
def _fetch_kg_requirements(kg_store: Any, project_dir: str) -> list[Any]:
    """Fetch requirements from KG and convert to dataclasses.

    Args:
        kg_store: KGDesignStore instance.
        project_dir: Project root directory for canonical slug resolution.

    Returns:
        List of Requirement objects from KG.
    """
    from orchestrata_ultimate.verification.traceability import Requirement

    slug = _project_slug(project_dir)
    kg_reqs = kg_store.get_requirements(slug)
    if not kg_reqs:
        return []

    reqs = [
        Requirement(
            id=r.get("req_id", ""),
            description=r.get("description", ""),
            story_id=r.get("story_id", ""),
            spec_id=r.get("spec_id", ""),
        )
        for r in kg_reqs
    ]
    logger.info("Loaded %d requirements from KG", len(reqs))
    return reqs


@traces(
    spec="026",
    story="Step 3 — RTM Generation",
    reqs=["REQ-026-014", "REQ-026-015", "REQ-026-016"],
)
def generate_rtm(
    trace_records: list[Any],
    requirements: list[Any],
    task_summary: str,
) -> Any:
    """Generate RTM from traces and requirements.

    Args:
        trace_records: TraceInfo objects from scan step.
        requirements: Requirement objects from KG.
        task_summary: Task description for project slug.

    Returns:
        RTMReport or None on failure.
    """
    try:
        from orchestrata_ultimate.verification.traceability import (
            generate_rtm as _gen_rtm,
        )

        slug = _task_slug(task_summary)
        rtm = _gen_rtm(trace_records, requirements, project=slug)
        if rtm.validation:
            v = rtm.validation
            logger.info(
                "RTM: %d verified, %d missing, %d orphan, %s%% coverage",
                v.total_verified,
                v.total_missing,
                v.total_orphan,
                v.coverage_percent,
            )
        return rtm
    except Exception as exc:
        logger.warning("RTM generation error: %s", exc)
        return None


@traces(
    spec="026",
    story="Step 4 — Code Sync",
    reqs=["REQ-026-017", "REQ-026-018", "REQ-026-019"],
)
def sync_code(
    kg_store: Any,
    project_dir: str,
    trace_records: list[Any],
) -> None:
    """Sync code trace records to knowledge graph.

    Args:
        kg_store: KGDesignStore instance, or None.
        project_dir: Project root directory for canonical KG project slug.
        trace_records: TraceInfo objects to sync.
    """
    if not kg_store:
        return
    try:
        slug = _project_slug(project_dir)
        dicts = [t.to_dict() if hasattr(t, "to_dict") else t for t in trace_records]
        kg_store.sync_code(slug, dicts)
        logger.info("Synced %d trace records to KG", len(dicts))
    except Exception as exc:
        logger.warning("Code sync error: %s", exc)


@traces(
    spec="026",
    story="Step 5 — Design Compliance Check",
    reqs=[
        "REQ-026-020",
        "REQ-026-021",
        "REQ-026-022",
        "REQ-026-023",
    ],
)
def check_compliance(kg_store: Any, project_dir: str) -> Any:
    """Check design vs code compliance via KG.

    Args:
        kg_store: KGDesignStore instance, or None.
        project_dir: Project root directory for canonical KG project slug.

    Returns:
        ComplianceResult or None if KG unavailable.
    """
    if not kg_store:
        return None
    try:
        slug = _project_slug(project_dir)
        result = kg_store.check_compliance(slug)
        if result:
            logger.info(
                "Design compliance: %s%% (%d missing, %d extra)",
                result.compliance_pct,
                len(result.missing_classes),
                len(result.extra_classes),
            )
        return result
    except Exception as exc:
        logger.warning("Compliance check error: %s", exc)
        return None


@traces(
    spec="026",
    story="Step 6 — RTM Persistence",
    reqs=[
        "REQ-026-024",
        "REQ-026-025",
    ],
)
def persist_rtm(
    kg_store: Any,
    rtm_report: Any,
    session_id: str,
    project_dir: str,
) -> dict[str, str]:
    """Persist RTM to disk and knowledge graph.

    Args:
        kg_store: KGDesignStore instance, or None.
        rtm_report: RTMReport to persist, or None.
        session_id: Current session ID for KG snapshot.
        project_dir: Project directory for disk output.

    Returns:
        Dict of file type to path for saved RTM files.
    """
    if not rtm_report:
        return {}
    files = _save_rtm_to_disk(rtm_report, project_dir)
    _save_rtm_to_kg(kg_store, rtm_report, session_id)
    return files


@traces(spec="026", story="Step 6 — RTM Persistence", reqs=["REQ-026-026"])
def _save_rtm_to_disk(
    rtm_report: Any,
    project_dir: str,
) -> dict[str, str]:
    """Write RTM report to disk as JSON and markdown.

    Args:
        rtm_report: RTMReport to save.
        project_dir: Project directory for docs output.

    Returns:
        Dict of file type to path.
    """
    if not project_dir:
        # An empty project_dir is a malformed/unset CXP packet field, NEVER
        # a legitimate "write to the current working directory" request --
        # os.path.join("", "docs") silently resolves to a CWD-relative
        # "docs" dir, which is how a stray test invocation (missing
        # project_dir in its packet payload) polluted THIS repo's own
        # tracked docs/rtm.{json,md} with a self-contradicting full-corpus
        # scan (2026-07 BDD-runtime-tracing audit). Refuse instead of
        # guessing a location; matches ``_project_slug``'s "unscoped"
        # sentinel discipline just above.
        logger.warning("RTM save skipped: no project_dir (would write under CWD)")
        return {}
    try:
        from orchestrata_ultimate.verification.traceability import save_rtm

        docs_dir = os.path.join(project_dir, "docs")
        files = save_rtm(rtm_report, docs_dir)
        logger.info("RTM saved: %s", files)
        return files
    except Exception as exc:
        logger.warning("RTM save error: %s", exc)
        return {}


@traces(
    spec="026",
    story="Step 6 — RTM Persistence",
    reqs=["REQ-026-027", "REQ-026-028"],
)
def _save_rtm_to_kg(
    kg_store: Any,
    rtm_report: Any,
    session_id: str,
) -> None:
    """Persist RTM snapshot to knowledge graph.

    Args:
        kg_store: KGDesignStore instance, or None.
        rtm_report: RTMReport with validation data.
        session_id: Session to attach snapshot to.
    """
    if not kg_store or not session_id:
        return
    try:
        from orchestrata_ultimate.knowledge_graph.kg_design import RTMSnapshot

        v = rtm_report.validation
        snapshot = RTMSnapshot(
            id=f"rtm-{uuid.uuid4().hex[:8]}",
            coverage_pct=(v.coverage_percent if v else 0.0),
            verified_count=(v.total_verified if v else 0),
            total_count=((v.total_verified + v.total_missing) if v else 0),
            missing_count=(v.total_missing if v else 0),
            orphan_count=(v.total_orphan if v else 0),
        )
        kg_store.close_session(
            session_id=session_id,
            achievements=[f"RTM: {snapshot.coverage_pct}% coverage"],
            rtm_snapshot=snapshot,
        )
        logger.info("RTM snapshot stored in KG")
    except Exception as exc:
        logger.warning("RTM KG storage error: %s", exc)