orchestrata_ultimate.verification.spec_completeness

Reference › orchestrata_ultimate › verification › spec_completeness

Answers "how much of this spec did the generated project implement?", for the QA volk. The entry point is evaluate_spec_completeness(project_dir, task_summary). It is re-exported by the package root.

The flow: _find_spec_checklist takes the first three-digit number in the task summary and looks for a spec directory with that prefix containing checklists/requirements.md; with no match the function returns an empty dict. Requirements are parsed into Prolog facts by the openspec requirements parser and into (id, description) tuples by _parse_req_tuples. implementation_scanner then matches them against AST evidence. _run_prolog_evaluation loads spec_completeness.pl, retracts earlier facts for the spec (so repeated calls in one process do not inflate the denominator), asserts the new facts and queries the coverage predicates. When Prolog is unavailable a pure Python fallback computes the same counts.

The result holds spec_coverage (for example "47/202 (23.3%)"), spec_complete, gaps (unresolved requirements are labelled as such), scan_degraded and scan_diagnostics. When scan_degraded is true an evidence source could not be read, so the gaps describe missing source and the coverage is a floor.

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/verification/spec_completeness.py

"""Spec completeness evaluation via Prolog.

Parses a spec's requirements checklist, scans generated code for
implementation evidence, asserts both as Prolog facts, and queries
the ``spec_completeness.pl`` rules to determine coverage and gaps.

This module is called by the QA volk after all review gates pass,
before declaring review complete.

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

import os
import re

from orchestrata_ultimate.core.infra_config import PROLOG_RULES_DIR
from orchestrata_ultimate.core.paths import resolve_project_root
from orchestrata_ultimate.openspec.requirements_parser import parse_requirements
from orchestrata_ultimate.verification.implementation_scanner import (
    match_requirements_to_evidence,
)

# Matches REQ lines to extract (req_id, description) tuples
_REQ_RE = re.compile(
    r"^-\s*\[[ x]\]\s*\*\*(?P<req_id>REQ-\d{3}-\d{3})\*\*:" r"\s*(?P<desc>.+)$"
)

# Bug fix (found while touching this file 2026-08-12): this counted SIX
# os.path.dirname() hops off __file__ -- verification -> orchestrata_ultimate
# -> src -> repo root -> workspace -> home -- landing two levels above the
# repo at `~/specs`, which never exists. `_find_spec_checklist` therefore
# always failed its `os.path.isdir(_SPECS_DIR)` check and every real call to
# `evaluate_spec_completeness` silently returned `{}`, in every environment,
# since this module was written. Four hops (verification -> orchestrata_ultimate
# -> src -> repo root) fixed it for the source tree; since the proper install
# (dd919d55c) the module sits in site-packages, where four hops is .venv/lib,
# so the root now comes from resolve_project_root (the nearest pyproject.toml).
_SPECS_DIR = str(resolve_project_root() / "specs")


def _find_spec_checklist(
    task_summary: str,
) -> tuple[str, str]:
    """Locate the spec checklist from the task summary.

    Searches the task summary for spec references (e.g. ``014``,
    ``015-plugin-sdk``) and resolves to the checklist path on disk.

    Args:
        task_summary: The task summary text from the CXP packet.

    Returns:
        Tuple of ``(spec_id, checklist_path)`` or ``("", "")``
        if no spec checklist is found.
    """
    if not os.path.isdir(_SPECS_DIR):
        return "", ""

    # Try to find a 3-digit spec number in the task summary
    match = re.search(r"(\d{3})", task_summary)
    if not match:
        return "", ""

    spec_num = match.group(1)
    for dirname in os.listdir(_SPECS_DIR):
        if dirname.startswith(spec_num):
            checklist = os.path.join(
                _SPECS_DIR,
                dirname,
                "checklists",
                "requirements.md",
            )
            if os.path.isfile(checklist):
                return spec_num, checklist

    return "", ""


def _parse_req_tuples(
    checklist_path: str,
) -> list[tuple[str, str]]:
    """Parse requirements as (req_id, description) tuples.

    Args:
        checklist_path: Absolute path to requirements.md.

    Returns:
        List of ``(req_id, description)`` tuples.
    """
    tuples: list[tuple[str, str]] = []
    with open(checklist_path, encoding="utf-8") as fh:
        for line in fh:
            stripped = line.strip()
            req_match = _REQ_RE.match(stripped)
            if req_match:
                tuples.append(
                    (
                        req_match.group("req_id"),
                        req_match.group("desc").strip(),
                    )
                )
    return tuples


def _run_prolog_evaluation(
    spec_id: str,
    req_facts: list[str],
    impl_facts: list[str],
) -> dict:
    """Run Prolog evaluation of spec completeness.

    Loads ``spec_completeness.pl``, asserts requirement and
    implementation facts, then queries coverage predicates.

    Args:
        spec_id: Spec number (e.g. ``"014"``).
        req_facts: List of ``spec_requirement/3`` fact strings.
        impl_facts: List of ``implemented/2`` fact strings.

    Returns:
        Dict with ``spec_complete``, ``implemented_count``,
        ``total_count``, and ``gaps`` list.
    """
    from orchestrata_ultimate.core.prolog_adapter import (
        PROLOG_AVAILABLE,
        PrologEngine,
    )

    if not PROLOG_AVAILABLE:
        return {}

    # janus-swi via the adapter: rules come from spec_completeness.pl (consulted)
    # and the requirement/implementation facts asserted below are ground.
    prolog = PrologEngine()

    rules_path = os.path.join(
        PROLOG_RULES_DIR,
        "spec_completeness.pl",
    )
    if os.path.isfile(rules_path):
        prolog.consult(rules_path)

    # Clear this spec's facts before asserting. spec_requirement/3 and
    # implemented/2 are dynamic and the engine outlives the call, so
    # without this a second evaluation of the same spec in one process
    # sees BOTH generations. Measured: evaluating one 1-requirement spec
    # twice reported total_count=2 -- coverage denominators grew with the
    # number of times the process had been asked, which is not a property
    # of the spec.
    for predicate in (
        f"spec_completeness:spec_requirement('{spec_id}', _, _)",
        f"spec_completeness:implemented('{spec_id}', _)",
    ):
        try:
            prolog.retractall(predicate)
        except Exception:
            continue

    # Assert requirement facts (module-qualified for SWI-Prolog)
    for fact in req_facts:
        try:
            prolog.assertz(f"spec_completeness:{fact}")
        except Exception:
            continue

    # Assert implementation facts (module-qualified for SWI-Prolog)
    for fact in impl_facts:
        try:
            prolog.assertz(f"spec_completeness:{fact}")
        except Exception:
            continue

    # Query spec_complete
    complete = bool(list(prolog.query(f"spec_complete('{spec_id}')")))

    # Query implementation_coverage
    impl_count = 0
    total_count = 0
    coverage_results = list(prolog.query(f"implementation_coverage('{spec_id}', I, T)"))
    if coverage_results:
        impl_count = int(coverage_results[0]["I"])
        total_count = int(coverage_results[0]["T"])

    # Query missing requirements individually to avoid
    # janus-swi PrologError on compound terms like gap(R, S).
    # Instead of gap_report/2 which returns findall(gap(...)),
    # we query missing_requirement/3 and collect results in Python.
    gaps: list[str] = []
    missing_results = list(
        prolog.query(f"missing_requirement('{spec_id}', ReqId, Desc)")
    )
    for result in missing_results:
        req_id = str(result.get("ReqId", ""))
        desc = str(result.get("Desc", ""))
        gaps.append(f"{req_id}: {desc}")

    return {
        "spec_complete": complete,
        "implemented_count": impl_count,
        "total_count": total_count,
        "gaps": gaps,
    }


def evaluate_spec_completeness(
    project_dir: str,
    task_summary: str,
) -> dict:
    """Evaluate spec completeness for a generated project.

    This is the main entry point called by the QA volk.  It:

    1. Finds the spec checklist from the task summary.
    2. Parses requirements into Prolog facts.
    3. Scans the project for implementation evidence.
    4. Runs Prolog evaluation.
    5. Returns a coverage report dict.

    Args:
        project_dir: Root directory of the generated project.
        task_summary: Task summary string (used to find spec).

    Returns:
        Dict with keys ``spec_coverage`` (str like ``"47/202"``),
        ``spec_complete`` (bool), ``gaps`` (list of str),
        ``scan_degraded`` (bool) and ``scan_diagnostics`` (list of str),
        or empty dict if no spec checklist is found.

        ``scan_degraded`` is the one that must not be ignored: when it is
        true an evidence source could not be read, so ``gaps`` describes
        the missing SOURCE rather than missing work, and the coverage
        number is a floor rather than a measurement.
    """
    spec_id, checklist_path = _find_spec_checklist(task_summary)
    if not spec_id:
        return {}

    # Parse requirements
    req_facts = parse_requirements(spec_id, checklist_path)
    if not req_facts:
        return {}

    req_tuples = _parse_req_tuples(checklist_path)

    # Scan implementation (AST-derived evidence) and match against requirements.
    # `unresolved` (a description with no identifier-shaped token at all) is
    # kept distinct from `gaps` (a named identifier with no matching evidence)
    # per the implementation-scanner honesty rule; both still count toward
    # "not implemented" in the coverage number below, which is unchanged in
    # kind -- only the gap text now says which is which.
    match_report = match_requirements_to_evidence(
        spec_id,
        req_tuples,
        project_dir,
    )
    impl_facts = match_report.implemented
    gaps = match_report.gaps
    unresolved = match_report.unresolved

    # A DEGRADED scan could not read some evidence source, so its gaps are
    # not findings -- they are the shape of the missing source. Reporting
    # them as ordinary gaps is precisely how commit 29bbf088 went
    # unnoticed: it deleted all non-Python evidence collection, and every
    # JS/TS-implemented requirement started reading as a clean gap.
    scan_diagnostics = list(match_report.diagnostics)

    # Try Prolog evaluation first
    prolog_result = _run_prolog_evaluation(
        spec_id,
        req_facts,
        impl_facts,
    )

    unresolved_strs = [
        f"{rid}: {desc} (unresolved — no structural identifier found)"
        for rid, desc in unresolved
    ]

    if prolog_result:
        impl_n = prolog_result["implemented_count"]
        total_n = prolog_result["total_count"]
        pct = (impl_n / total_n * 100) if total_n > 0 else 0.0
        gap_strs = prolog_result["gaps"]
        # Enrich gaps with req IDs from Python-side matching
        if not gap_strs and (gaps or unresolved):
            gap_strs = [f"{rid}: {desc}" for rid, desc in gaps] + unresolved_strs
        return {
            "spec_coverage": (f"{impl_n}/{total_n} ({pct:.1f}%)"),
            "spec_complete": prolog_result["spec_complete"],
            "gaps": gap_strs,
            "scan_degraded": bool(scan_diagnostics),
            "scan_diagnostics": scan_diagnostics,
        }

    # Python fallback when Prolog is unavailable
    total = len(req_tuples)
    implemented = total - len(gaps) - len(unresolved)
    pct = (implemented / total * 100) if total > 0 else 0.0
    gap_strs = [f"{rid}: {desc}" for rid, desc in gaps] + unresolved_strs
    return {
        "spec_coverage": (f"{implemented}/{total} ({pct:.1f}%)"),
        "spec_complete": len(gaps) == 0 and len(unresolved) == 0,
        "gaps": gap_strs,
        "scan_degraded": bool(scan_diagnostics),
        "scan_diagnostics": scan_diagnostics,
    }