orchestrata_ultimate.knowledge_graph.requirement_resolution

Reference › orchestrata_ultimate › knowledge_graph › requirement_resolution

Canonical requirement identity resolution for the operational graph (orchkg on orchgraph). The graph models requirements as a chain, (:Spec)-[:DEFINES_REQUIREMENT]->(:Requirement)-[:HAS_ITEM]->(:RequirementItem), and code symbols trace to RequirementItem nodes. This module resolves those items back to their parent requirement and owning spec so every consumer sees one logical record. The frozen dataclass RequirementChain holds the unified id, a description, the owning spec ids, the item ids and a graph distance. resolve_requirement_chain takes seed item ids, resolve_requirements_for_symbol starts from a symbol name, and closure_requirement_chain expands outward to a bounded depth (default 2) and result count (default 20). All three read through a synchronous graph repository and accept an optional project id; _canonical_id and _canonical_description normalise whichever record shape comes back. They only read. context_injector and some tools import them; label and relationship names come from canonical_schema.

At a glance

Class diagram

Dependency diagram

API

Classes

RequirementChain (dataclass)

Decorators: dataclass(frozen=True) · Line 37

Unified Requirement + RequirementItem + Spec record.

Functions

Source

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

"""Canonical requirement identity resolution.

Unifies ``Requirement`` and ``RequirementItem`` so every KG consumer treats
them as one logical chain:

    (:Spec)-[:DEFINES_REQUIREMENT]->(:Requirement)-[:HAS_ITEM]->(:RequirementItem)
                                      ^                      ^
                                      |                      |
                         unified RequirementChain id          item_ids

Code symbols trace to ``RequirementItem`` nodes. The helpers in this module
resolve those items to their parent ``Requirement`` (and owning ``Spec``),
returning a stable, canonical record regardless of whether the seed was an
item id or a requirement id.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.canonical_schema import (
    LABEL_REQUIREMENT,
    LABEL_REQUIREMENT_ITEM,
    REL_DEFINES_REQUIREMENT,
    REL_HAS_ITEM,
    REL_REQUIRES,
    REL_TRACES_TO,
)

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph.interfaces import SyncGraphRepository


@dataclass(frozen=True)
class RequirementChain:
    """Unified Requirement + RequirementItem + Spec record.

    Attributes:
        id: Canonical requirement id. If the resolved record has a parent
            ``Requirement``, this is the parent's id; otherwise it falls back
            to the ``RequirementItem`` id.
        description: Human-readable requirement text, coalesced from
            ``title``, ``description``, ``text``, or ``name`` properties.
        specs: List of owning Spec ids/natural ids.
        item_ids: The RequirementItem ids that trace to this requirement.
        distance: Hop distance from the seed (0 = direct trace target).
    """

    id: str
    description: str
    specs: list[str] = field(default_factory=list)
    item_ids: list[str] = field(default_factory=list)
    distance: int = 0


@traces(spec="106", reqs=["REQ-106-001"])
def resolve_requirement_chain(
    seed_item_ids: list[str],
    repo: SyncGraphRepository,
    project_id: str | None = None,
) -> list[RequirementChain]:
    """Resolve RequirementItem ids to unified RequirementChain records.

    For each seed item id, the query walks ``(:RequirementItem)<-[:HAS_ITEM]-
    (:Requirement)<-[:DEFINES_REQUIREMENT]-(:Spec)``. If no parent Requirement
    exists, the item itself is returned as a single-element chain.

    Args:
        seed_item_ids: RequirementItem ids to resolve.
        repo: Operational graph repository.
        project_id: Optional project scope filter.

    Returns:
        A deduplicated list of :class:`RequirementChain` records, one per
        canonical requirement id.
    """
    if not seed_item_ids:
        return []

    params: dict = {"item_ids": seed_item_ids}
    project_filter = ""
    if project_id is not None:
        project_filter = " AND ri.project_id = $project_id"
        params["project_id"] = project_id

    rows = repo.execute_read(
        "MATCH (ri:RequirementItem) WHERE ri.id IN $item_ids " + project_filter + " "
        f"OPTIONAL MATCH (ri)<-[:{REL_HAS_ITEM}]-(r:Requirement) "
        f"OPTIONAL MATCH (r)<-[:{REL_DEFINES_REQUIREMENT}]-(s:Spec) "
        "RETURN coalesce(r.id, ri.id) AS id, "
        "       coalesce(r.title, r.description, r.id, "
        "                ri.text, ri.description, ri.name, ri.id) AS description, "
        "       collect(DISTINCT coalesce(s.id, s.natural_id)) AS specs, "
        "       collect(DISTINCT ri.id) AS item_ids",
        params,
    )

    # The query aggregates per canonical id (implicit GROUP BY on the
    # non-aggregate RETURN expressions ``id``/``description``) and
    # ``Requirement.id`` carries a UNIQUE constraint, so it emits exactly one
    # row per canonical requirement id. No row-level merge is needed — and no
    # merge could ever fire, since two rows sharing a canonical id would also
    # share its (deterministically-derived) description and thus collapse into
    # one row before reaching this loop.
    seen: dict[str, RequirementChain] = {}
    for row in rows:
        req_id = row.get("id")
        if not req_id:
            continue
        seen[req_id] = RequirementChain(
            id=req_id,
            description=row.get("description") or "",
            specs=[s for s in (row.get("specs") or []) if s],
            item_ids=[i for i in (row.get("item_ids") or []) if i],
        )

    return list(seen.values())


@traces(spec="106", reqs=["REQ-106-001"])
def resolve_requirements_for_symbol(
    symbol_name: str,
    repo: SyncGraphRepository,
    project_id: str | None = None,
) -> list[RequirementChain]:
    """Resolve requirements traced to a code symbol by name.

    Matches ``(sym)-[:TRACES_TO]->(:RequirementItem)`` and delegates to
    :func:`resolve_requirement_chain` for unification with parent Requirements.

    Args:
        symbol_name: Class/function name to look up.
        repo: Operational graph repository.
        project_id: Optional project scope filter.

    Returns:
        Unified requirement chains for the symbol.
    """
    params: dict = {"name": symbol_name}
    project_filter = ""
    if project_id is not None:
        project_filter = " AND ri.project_id = $project_id"
        params["project_id"] = project_id

    # One row per traced item (no aggregation): a symbol with no TRACES_TO
    # edges yields zero rows, so the ``if not rows`` guard below is the real
    # no-match path — not a dead defensive branch.
    rows = repo.execute_read(
        f"MATCH (sym)-[:{REL_TRACES_TO}]->(ri:RequirementItem) "
        "WHERE sym.name = $name " + project_filter + " "
        "RETURN DISTINCT ri.id AS item_id",
        params,
    )
    if not rows:
        return []
    item_ids = [str(item_id) for r in rows if (item_id := r.get("item_id"))]
    return resolve_requirement_chain(item_ids, repo, project_id=project_id)


@traces(spec="106", reqs=["REQ-106-001"])
def closure_requirement_chain(
    seed_item_ids: list[str],
    repo: SyncGraphRepository,
    project_id: str | None = None,
    max_depth: int = 2,
    max_results: int = 20,
) -> list[RequirementChain]:
    """Walk the requirement hierarchy around the seed items.

    Starting from ``RequirementItem`` ids, the closure collects:

    * distance 0 — the seed items themselves (unified to their parent
      ``Requirement`` when one exists).
    * distance 1+ — ``Requirement`` nodes linked by ``REQUIRES`` up to
      ``max_depth``.

    Results are deduplicated by canonical requirement id and capped at
    ``max_results``. The ``distance`` field records the shortest hop count
    from any seed.

    Args:
        seed_item_ids: RequirementItem ids that are the starting points.
        repo: Operational graph repository.
        project_id: Optional project scope filter.
        max_depth: Maximum relationship hops to walk from the seeds.
        max_results: Hard cap on returned chains.

    Returns:
        A list of :class:`RequirementChain` records, shortest distance first.
    """
    if not seed_item_ids:
        return []

    # Phase 1: resolve seed items to their parent Requirements (distance 0).
    direct = resolve_requirement_chain(seed_item_ids, repo, project_id=project_id)
    seen: dict[str, RequirementChain] = {}
    for chain in direct:
        seen[chain.id] = RequirementChain(
            id=chain.id,
            description=chain.description,
            specs=chain.specs,
            item_ids=chain.item_ids,
            distance=0,
        )

    # Phase 2: walk REQUIRES edges from the direct Requirements.
    if not seen:
        return []

    root_ids = list(seen)
    params: dict = {"root_ids": root_ids}
    project_filter = ""
    if project_id is not None:
        project_filter = " AND r.project_id = $project_id"
        params["project_id"] = project_id

    # Variable-length path bounds must be literals, so interpolate max_depth.
    rows = repo.execute_read(
        "MATCH (r:Requirement) WHERE r.id IN $root_ids " + project_filter + " "
        f"OPTIONAL MATCH path = (r)-[:{REL_REQUIRES}*1..{max_depth}]->(dep:Requirement) "
        "RETURN dep.id AS id, dep.title AS title, dep.description AS description, "
        "       length(path) AS hops",
        params,
    )

    for row in rows:
        dep_id = row.get("id")
        if not dep_id or dep_id in seen:
            continue
        description = row.get("title") or row.get("description") or dep_id or ""
        hops = row.get("hops") or 1
        seen[dep_id] = RequirementChain(
            id=dep_id,
            description=str(description),
            distance=hops,
        )

    return sorted(seen.values(), key=lambda c: (c.distance, c.id))[:max_results]


def _canonical_id(record: dict, label: str) -> str | None:
    """Extract the canonical id from a requirement-ish record.

    Handles both the canonical ``id`` property and the legacy ``req_id`` /
    ``item_id`` properties that some writers still emit.
    """
    if label == LABEL_REQUIREMENT:
        return record.get("id") or record.get("req_id")
    if label == LABEL_REQUIREMENT_ITEM:
        return record.get("id") or record.get("item_id")
    return record.get("id")


def _canonical_description(record: dict) -> str:
    """Extract the most human-readable description available."""
    for key in ("title", "description", "text", "name", "statement"):
        value = record.get(key)
        if value:
            return str(value)
    return ""