orchestrata_ultimate.agents.volks.compliance.volk

Reference › orchestrata_ultimate › agents › volks › compliance › volk

ComplianceVolk (alias U:Cp, role Compliance) is the terminal volk of the pipeline and the only one that makes no LLM call (specification 026). It checks design against code and builds the requirements traceability matrix (RTM) from @traces markers. It extends BaseAgent (base) and delegates the pipeline steps to _pipeline; the package __init__ (init) re-exports it for the kernel pipeline. ComplianceAgent is an alias subclass.

What it receives: a compliance packet (typically from QA, or from the Documentor so generated docs are traced) carrying the project directory, task summary and session id. What it produces: a final packet addressed to HUMAN with subject compliance_complete on a pass or compliance_failed on a violation, whose payload holds the RTM, the compliance result, the RTM file paths, both percentages and a readable report. A failure is therefore a first-class terminal outcome that the kernel reports as an unsuccessful completion.

process calls _handle_compliance, which stores the packet fields (_extract_packet_fields) and runs _run_pipeline: it initialises the knowledge-graph store lazily (_init_kg), scans the src and tests directories for traces, loads requirements from the graph, generates the RTM, syncs code nodes to the graph, checks design against code and persists the RTM to disk and graph. _calculate_results takes the RTM coverage and design-compliance percentages (_extract_coverage, _extract_compliance_pct) and passes the run when the coverage meets the volk's compliance threshold. _build_report_artifact and build_report_lines (with _rtm_section, _compliance_section, _files_section) produce the human-readable report, and _build_response assembles the outgoing packet.

At a glance

Class diagram

Dependency diagram

API

Classes

ComplianceVolk (pydantic)

Bases: BaseAgent · Line 46

Compliance checker volk -- deterministic, no LLM.

ComplianceAgent (pydantic)

Bases: ComplianceVolk · Line 458

Agent alias for ComplianceVolk.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/agents/volks/compliance/volk.py

# mypy: ignore-errors
"""U:Cp -- Compliance volk for design-vs-code checking and RTM.

Deterministic (no LLM calls). Runs after U:Qa completes:
1. Scans @traces in source and test code
2. Loads requirements from KG
3. Generates RTM
4. Syncs code nodes to KG
5. Checks design vs code compliance
6. Persists RTM to disk and KG
7. Produces compliance report
"""

import logging
import os
from typing import Any

from pydantic import Field

from orchestrata_ultimate.agents.volks.base import BaseAgent
from orchestrata_ultimate.agents.volks.compliance._pipeline import (
    _task_slug,
    check_compliance,
    generate_rtm,
    load_requirements,
    persist_rtm,
    scan_traces,
    sync_code,
)
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus

logger = logging.getLogger(__name__)

# Re-exported so existing imports from this module keep working.
__all__ = [
    "ComplianceVolk",
    "_task_slug",
    "_extract_coverage",
    "_extract_compliance_pct",
    "_build_payload",
    "build_report_lines",
]


class ComplianceVolk(BaseAgent):
    """Compliance checker volk -- deterministic, no LLM.

    Runs a 7-step pipeline that scans @traces decorators,
    loads requirements from the knowledge graph, generates an
    RTM, checks design-vs-code compliance, and produces a
    human-readable report.
    """

    alias: str = Field(default="U:Cp")
    role: str = "Compliance"

    def __init__(self) -> None:
        """Initialize compliance volk with default thresholds."""
        super().__init__()
        self.escalation_target: str | None = None
        self.project_dir: str = ""
        self.task_summary: str = ""
        self.kg_store: Any = None
        self.compliance_threshold: float = 0.0

    @traces(
        spec="026",
        story="KG Lazy Initialisation",
        reqs=["REQ-026-033", "REQ-026-034"],
    )
    def _init_kg(self) -> None:
        """Lazy-init KG connection.

        Uses orchestrata.knowledge.kg_design.KGDesignStore.
        Non-fatal: sets kg_store to None on failure.
        """
        if self.kg_store is not None:
            return
        try:
            from orchestrata_ultimate.knowledge_graph.kg_design import KGDesignStore

            self.kg_store = KGDesignStore()
        except Exception as exc:
            logger.warning("KG init failed: %s", exc)
            self.kg_store = None

    @traces(
        spec="026",
        story="Pipeline Orchestration",
        reqs=["REQ-026-001"],
    )
    async def process(self, packet: CXPPacket) -> CXPPacket | None:
        """Route incoming CXP packets to compliance handler.

        Args:
            packet: Incoming CXP packet to process.

        Returns:
            Response CXP packet, or None if subject is unhandled.
        """
        if packet.subject == "compliance_check":
            return self._handle_compliance(packet)
        return None

    @traces(
        spec="026",
        story="Pipeline Orchestration",
        reqs=["REQ-026-002"],
    )
    def _handle_compliance(self, packet: CXPPacket) -> CXPPacket:
        """Run full compliance pipeline.

        Args:
            packet: CXP packet with compliance_check subject.

        Returns:
            CXP response packet with compliance results.
        """
        self._extract_packet_fields(packet)
        report = self._run_pipeline(packet)
        return self._build_response(packet, report)

    @traces(
        spec="026",
        story="Pipeline Orchestration",
        reqs=["REQ-026-004"],
    )
    def _extract_packet_fields(self, packet: CXPPacket) -> None:
        """Extract and store fields from incoming packet.

        Args:
            packet: Incoming CXP packet.
        """
        self.status = WorkStatus.IN_PROGRESS
        self.current_thread = packet.thread_id
        self.project_dir = packet.payload.get("project_dir", "")
        self.task_summary = packet.payload.get("task_summary", "")
        self.summary = "Running compliance checks"
        logger.info("Compliance check: %s", self.project_dir)

    @traces(
        spec="026",
        story="Pipeline Orchestration",
        reqs=["REQ-026-003"],
    )
    def _run_pipeline(self, packet: CXPPacket) -> dict[str, Any]:
        """Execute the 7-step compliance pipeline.

        Args:
            packet: CXP packet containing session_id.

        Returns:
            Report dict with rtm, compliance, files, and metrics.
        """
        session_id = packet.payload.get("session_id", "")
        src_dir = os.path.join(self.project_dir, "src")
        tests_dir = os.path.join(self.project_dir, "tests")

        self._init_kg()
        trace_records = scan_traces(src_dir, tests_dir)
        # KG lookups/writes scope by project_dir (the canonical KG project
        # slug), NOT task_summary -- a free-text task/intent summary is
        # neither stable nor unique per project (2026-07 audit: it produced
        # empty/truncated anonymous Project hub nodes). generate_rtm's
        # `task_summary` argument is a display-only label, unaffected.
        requirements = load_requirements(self.kg_store, self.project_dir)
        rtm_report = generate_rtm(trace_records, requirements, self.task_summary)
        sync_code(self.kg_store, self.project_dir, trace_records)
        compliance_result = check_compliance(self.kg_store, self.project_dir)
        rtm_files = persist_rtm(
            self.kg_store,
            rtm_report,
            session_id,
            self.project_dir,
        )

        report = self._calculate_results(rtm_report, compliance_result, rtm_files)
        self._build_report_artifact(report)
        return report

    @traces(
        spec="026",
        story="Report Production",
        reqs=["REQ-026-029", "REQ-026-030"],
    )
    def _calculate_results(
        self,
        rtm_report: Any,
        compliance_result: Any,
        rtm_files: dict[str, str],
    ) -> dict[str, Any]:
        """Calculate coverage and pass/fail from pipeline outputs.

        Args:
            rtm_report: RTMReport from step 3, or None.
            compliance_result: ComplianceResult from step 5, or None.
            rtm_files: File paths from step 6.

        Returns:
            Report dict with all metrics.
        """
        coverage_pct = _extract_coverage(rtm_report)
        compliance_pct = _extract_compliance_pct(compliance_result)
        passed = coverage_pct >= self.compliance_threshold

        return {
            "rtm": rtm_report,
            "compliance": compliance_result,
            "rtm_files": rtm_files,
            "coverage_pct": coverage_pct,
            "compliance_pct": compliance_pct,
            "passed": passed,
        }

    @traces(
        spec="026",
        story="Pipeline Orchestration",
        reqs=["REQ-026-004", "REQ-026-005"],
    )
    def _build_response(
        self,
        packet: CXPPacket,
        report: dict[str, Any],
    ) -> CXPPacket:
        """Build the response CXP packet from report.

        A compliance violation (``passed=False``) is a first-class terminal
        FAILED pipeline outcome: the response carries the ``compliance_failed``
        subject (so the kernel emits ``emit_complete(success=False, ...)``) and
        surfaces the full human-readable compliance report in the payload. A
        clean pass keeps the ``compliance_complete`` subject.

        Args:
            packet: Original incoming packet.
            report: Calculated report dict.

        Returns:
            CXP response packet — ``compliance_complete`` on pass,
            ``compliance_failed`` on violation.
        """
        passed = report["passed"]
        coverage_pct = report["coverage_pct"]
        compliance_pct = report["compliance_pct"]

        status_str = "PASS" if passed else "FAIL"
        logger.info(
            "Compliance %s: RTM coverage=%s%%, design compliance=%s%%",
            status_str,
            coverage_pct,
            compliance_pct,
        )

        # RESOLVED is the only terminal work status available on the packet
        # state machine; the FAILED *pipeline* outcome is signalled by the
        # ``compliance_failed`` subject + the ``passed`` flag in the payload,
        # which the kernel turns into emit_complete(success=False).
        self.status = WorkStatus.RESOLVED
        self.summary = (
            f"Compliance {status_str}: {coverage_pct}% RTM, {compliance_pct}% design"
        )

        report_text = "\n".join(build_report_lines(report))
        payload = _build_payload(report, self.project_dir, self.task_summary)
        payload["report"] = report_text

        subject = "compliance_complete" if passed else "compliance_failed"
        return CXPPacket(
            from_agent="U:Cp",
            to_agent="HUMAN",
            thread_id=packet.thread_id,
            subject=subject,
            payload=payload,
            work_status=WorkStatus.RESOLVED,
        )

    @traces(
        spec="026",
        story="Report Production",
        reqs=["REQ-026-032"],
    )
    def _build_report_artifact(
        self,
        report: dict[str, Any],
    ) -> None:
        """Build human-readable compliance report artifact.

        Args:
            report: Full report dict from pipeline.
        """
        lines = build_report_lines(report)
        self.output_artifacts.append("\n".join(lines))


@traces(spec="026", story="Report Production", reqs=["REQ-026-029"])
def _extract_coverage(rtm_report: Any) -> float:
    """Extract coverage percentage from RTM report.

    Args:
        rtm_report: RTMReport or None.

    Returns:
        Coverage percentage, 0.0 if unavailable.
    """
    if not rtm_report:
        return 0.0
    if not hasattr(rtm_report, "validation"):
        return 0.0
    v = rtm_report.validation
    if not v:
        return 0.0
    return v.coverage_percent


@traces(spec="026", story="Report Production", reqs=["REQ-026-029"])
def _extract_compliance_pct(compliance_result: Any) -> float:
    """Extract compliance percentage from result.

    Args:
        compliance_result: ComplianceResult or None.

    Returns:
        Compliance percentage, 0.0 if unavailable.
    """
    if not compliance_result:
        return 0.0
    return compliance_result.compliance_pct


@traces(
    spec="026",
    story="Pipeline Orchestration",
    reqs=["REQ-026-004", "REQ-026-005"],
)
def _build_payload(
    report: dict[str, Any],
    project_dir: str,
    task_summary: str,
) -> dict[str, Any]:
    """Build the response payload dict.

    Args:
        report: Calculated report dict.
        project_dir: Project directory path.
        task_summary: Task description string.

    Returns:
        Payload dict for CXP response packet.
    """
    rtm_report = report.get("rtm")
    compliance_result = report.get("compliance")
    return {
        "passed": report["passed"],
        "coverage_pct": report["coverage_pct"],
        "compliance_pct": report["compliance_pct"],
        "rtm_files": report["rtm_files"],
        "missing_classes": (
            compliance_result.missing_classes if compliance_result else []
        ),
        "missing_reqs": (
            rtm_report.validation.missing
            if rtm_report and rtm_report.validation
            else []
        ),
        "project_dir": project_dir,
        "task_summary": task_summary,
    }


@traces(
    spec="026",
    story="Report Production",
    reqs=["REQ-026-031"],
)
def build_report_lines(report: dict[str, Any]) -> list[str]:
    """Build human-readable compliance report lines.

    Args:
        report: Full report dict from pipeline.

    Returns:
        List of report lines.
    """
    lines = ["=== COMPLIANCE REPORT ===", ""]
    lines.extend(_rtm_section(report.get("rtm")))
    lines.extend(_compliance_section(report.get("compliance")))
    lines.extend(_files_section(report.get("rtm_files", {})))

    status = "PASS" if report.get("passed") else "FAIL"
    lines.append(f"## Result: {status}")
    return lines


@traces(spec="026", story="Report Production", reqs=["REQ-026-031"])
def _rtm_section(rtm: Any) -> list[str]:
    """Build RTM coverage section of the report.

    Args:
        rtm: RTMReport or None.

    Returns:
        Lines for RTM section.
    """
    if not rtm or not hasattr(rtm, "validation") or not rtm.validation:
        return []
    v = rtm.validation
    return [
        "## RTM Coverage",
        f"  Verified:  {v.total_verified}",
        f"  Missing:   {v.total_missing}",
        f"  Orphan:    {v.total_orphan}",
        f"  Coverage:  {v.coverage_percent}%",
        "",
    ]


@traces(spec="026", story="Report Production", reqs=["REQ-026-031"])
def _compliance_section(compliance: Any) -> list[str]:
    """Build design compliance section of the report.

    Args:
        compliance: ComplianceResult or None.

    Returns:
        Lines for compliance section.
    """
    if not compliance:
        return []
    return [
        "## Design Compliance",
        f"  Compliance: {compliance.compliance_pct}%",
        f"  Missing classes: {compliance.missing_classes}",
        f"  Extra classes: {compliance.extra_classes}",
        "",
    ]


@traces(spec="026", story="Report Production", reqs=["REQ-026-031"])
def _files_section(rtm_files: dict[str, str]) -> list[str]:
    """Build RTM files section of the report.

    Args:
        rtm_files: Dict of file type to path.

    Returns:
        Lines for files section.
    """
    if not rtm_files:
        return []
    lines = ["## RTM Files"]
    for ftype, fpath in rtm_files.items():
        lines.append(f"  {ftype}: {fpath}")
    lines.append("")
    return lines


# Class alias for orchestrata-ultimate naming
class ComplianceAgent(ComplianceVolk):
    """Agent alias for ComplianceVolk."""

    pass