orchestrata_ultimate.knowledge_graph.context_pack.relevance

Reference › orchestrata_ultimate › knowledge_graph › context_pack › relevance

Symbolic relevance refinement of a selected subgraph using Prolog (SPEC-106 FR-006). The ranking in selector is a linear score over independent per-node signals, so it cannot express relational relevance such as "this node shares a requirement with what was asked for". KgRelevanceRetriever adds that. It is built from a PrologBridge and a synchronous graph repository, with a recent-change window. refine projects the selected subgraph's signals, and the Tier-H runtime signals for those node ids read from the operational graph (orchkg on orchgraph), as ground Prolog facts. The task-type rules in prolog/kg_relevance.pl then deduce a relevance tier per node, and the facts are retracted after each call so none leak across requests (REQ-106-064). rescore_by_tier re-ranks the subgraph by a discrete tier bonus while preserving the FR-003 order within a tier. Per the doctrine in its docstring, variable-bearing rules live in the .pl file and Python asserts only ground facts. The layer is opt-in: without a retriever the FR-003 ordering is unchanged (REQ-106-063). The kernel pipeline and context_injector import it.

At a glance

Class diagram

Dependency diagram

API

Classes

KgRelevanceRetriever

Decorators: traces(spec='106', reqs=['REQ-106-060', 'REQ-106-061', 'REQ-106-064', 'REQ-106-065']) · Line 113

Deduces a :data:RelevanceTier per selected node via kg_relevance.pl.

Functions

Source

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

"""Symbolic (Prolog) relevance refinement of a selected subgraph (SPEC-106 FR-006).

FR-003's :func:`~orchestrata_ultimate.knowledge_graph.context_pack.selector.select`
ranks nodes by a fixed linear score over INDEPENDENT per-node signals. It cannot
express *relational* relevance — "this node shares a requirement with what you
asked for", "this node is a definition contained by a linked root". This module
adds that, symbolically and deterministically, by projecting the already-selected
subgraph's graph signals as GROUND Prolog facts and letting the task-type rules in
``prolog/kg_relevance.pl`` deduce a :data:`RelevanceTier` per node.

Doctrine (project-wide): variable-bearing rules live in the ``.pl`` file, consulted;
Python asserts ONLY ground facts, and retracts them after each refinement so no
facts leak across requests (REQ-106-064). The whole layer is opt-in: with no
:class:`KgRelevanceRetriever` wired, callers keep the FR-003-only ordering
byte-for-byte (REQ-106-063).
"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.canonical_schema import (
    REL_DEFINES,
    REL_HAS_ITEM,
    REL_MENTIONS,
    REL_TRACES_TO,
)
from orchestrata_ultimate.knowledge_graph.context_pack.selector import (
    RankedNode,
    RankedSubgraph,
)

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph import SyncGraphRepository
    from orchestrata_ultimate.infra.prolog_bridge import PrologBridge

# The RelevanceTier vocabulary (REQ-106-061). Ordered most- to least-relevant.
TIER_CORE = "core"
TIER_RELATED = "related"
TIER_PERIPHERAL = "peripheral"

# Discrete tier bonus added to the FR-003 rank score (REQ-106-062). The gap
# between tiers (1000.0) deliberately exceeds the maximum spread the FR-003
# linear score can produce at repo scale (hop/trace/tested/recency/ambiguity
# weights sum to well under 1000), so tier ALWAYS stratifies first while the
# FR-003 score orders WITHIN a tier and the node-id tie-break is preserved.
_TIER_BONUS = {TIER_CORE: 2000.0, TIER_RELATED: 1000.0, TIER_PERIPHERAL: 0.0}

# The ruleset ships beside the other governance rules so PrologBridge's rules
# dir picks it up; we also load it explicitly (idempotent) so a bridge that was
# never told to load-all still works.
_RULESET = Path(__file__).resolve().parents[2] / "prolog" / "kg_relevance.pl"

# The dynamic predicates the retriever projects — retracted before AND after
# every refinement (REQ-106-064).
_DYNAMIC_PREDICATES = (
    ("kg_candidate", 3),
    ("kg_root", 1),
    ("kg_traces", 2),
    ("kg_tested", 1),
    ("kg_defines", 2),
    ("kg_task", 1),
    ("kg_ambiguous", 1),
    ("kg_recent", 1),
    ("kg_failed", 1),
)

# Default lookback for runtime "recent" signals (seconds).
_DEFAULT_RECENT_WINDOW = 30 * 60


def _atom(value: str) -> str:
    """Quote a Python string as a Prolog atom (ids carry ``::``/``-``/``.``)."""
    escaped = value.replace("\\", "\\\\").replace("'", "\\'")
    return f"'{escaped}'"


@traces(spec="106", reqs=["REQ-106-062"])
def rescore_by_tier(subgraph: RankedSubgraph, tiers: dict[str, str]) -> RankedSubgraph:
    """Re-rank ``subgraph`` by discrete tier bonus, FR-003 order preserved within.

    Pure and deterministic: a node absent from ``tiers`` is treated as
    ``peripheral`` (zero bonus), so an empty ``tiers`` returns an
    order-identical subgraph (REQ-106-063). Ties break on node id exactly as
    FR-003 does (REQ-106-062).
    """
    rescored = [
        RankedNode(
            node_id=n.node_id,
            label=n.label,
            properties=n.properties,
            template=n.template,
            hop_distance=n.hop_distance,
            trace_degree=n.trace_degree,
            tested_by=n.tested_by,
            ambiguous=n.ambiguous,
            rank_score=n.rank_score
            + _TIER_BONUS.get(tiers.get(n.node_id, TIER_PERIPHERAL), 0.0),
        )
        for n in subgraph.nodes
    ]
    rescored.sort(key=lambda rn: (-rn.rank_score, rn.node_id))
    return RankedSubgraph(task_type=subgraph.task_type, nodes=tuple(rescored))


@traces(
    spec="106",
    reqs=["REQ-106-060", "REQ-106-061", "REQ-106-064", "REQ-106-065"],
)
class KgRelevanceRetriever:
    """Deduces a :data:`RelevanceTier` per selected node via ``kg_relevance.pl``.

    Uses the SAME worker-owned SWI engine as the rest of the system (through the
    injected :class:`PrologBridge`), never a private engine, so the thread-
    confined Prolog affinity is respected.
    """

    def __init__(
        self,
        bridge: PrologBridge,
        repo: SyncGraphRepository,
        *,
        recent_window_seconds: int = _DEFAULT_RECENT_WINDOW,
    ) -> None:
        self._bridge = bridge
        self._repo = repo
        self._recent_window_seconds = recent_window_seconds
        self._loaded = False

    def _ensure_loaded(self) -> None:
        if not self._loaded:
            self._bridge.load_rules(_RULESET)
            self._loaded = True

    def _clear(self) -> None:
        for name, arity in _DYNAMIC_PREDICATES:
            wildcards = ",".join("_" for _ in range(arity))
            self._bridge.query(f"retractall({name}({wildcards}))")

    def _project(self, subgraph: RankedSubgraph) -> None:
        node_ids = [n.node_id for n in subgraph.nodes]
        self._bridge.query(f"assertz(kg_task({_atom(subgraph.task_type.value)}))")
        for n in subgraph.nodes:
            self._bridge.query(
                f"assertz(kg_candidate({_atom(n.node_id)},{_atom(n.label)},"
                f"{n.hop_distance}))"
            )
            if n.hop_distance == 0:
                self._bridge.query(f"assertz(kg_root({_atom(n.node_id)}))")
            if n.tested_by:
                self._bridge.query(f"assertz(kg_tested({_atom(n.node_id)}))")
            if n.ambiguous:
                self._bridge.query(f"assertz(kg_ambiguous({_atom(n.node_id)}))")

        # Relational facts the RankedNode does not carry: TRACES_TO targets,
        # their parent Requirements (via HAS_ITEM), and DEFINES edges, all
        # restricted to the selected node set. Projecting parent Requirement ids
        # unifies Requirement/RequirementItem identity for requirement-sharing
        # logic: a symbol tracing to a RequirementItem shares a requirement with
        # a symbol tracing to the parent Requirement.
        traces_rows = self._repo.execute_read(
            f"MATCH (n)-[:{REL_TRACES_TO}]->(r) WHERE n.id IN $ids "
            "RETURN n.id AS nid, r.id AS rid",
            {"ids": node_ids},
        )
        for row in traces_rows:
            if row.get("nid") and row.get("rid"):
                self._bridge.query(
                    f"assertz(kg_traces({_atom(row['nid'])},{_atom(row['rid'])}))"
                )

        parent_req_rows = self._repo.execute_read(
            f"MATCH (n)-[:{REL_TRACES_TO}]->(ri:RequirementItem)<-[:{REL_HAS_ITEM}]-(r:Requirement) "
            "WHERE n.id IN $ids "
            "RETURN n.id AS nid, r.id AS rid",
            {"ids": node_ids},
        )
        for row in parent_req_rows:
            if row.get("nid") and row.get("rid"):
                self._bridge.query(
                    f"assertz(kg_traces({_atom(row['nid'])},{_atom(row['rid'])}))"
                )

        defines_rows = self._repo.execute_read(
            f"MATCH (p)-[:{REL_DEFINES}]->(c) WHERE p.id IN $ids AND c.id IN $ids "
            "RETURN p.id AS pid, c.id AS cid",
            {"ids": node_ids},
        )
        for row in defines_rows:
            if row.get("pid") and row.get("cid"):
                self._bridge.query(
                    f"assertz(kg_defines({_atom(row['pid'])},{_atom(row['cid'])}))"
                )

    def _project_runtime_signals(self, node_ids: list[str]) -> None:
        """Project Tier-H runtime signal facts for the selected node ids.

        ``kg_recent/1`` is asserted for any node mentioned by a Tier-H signal
        (message, gateway event, decision, tool call) within the configured
        recent window. ``kg_failed/1`` is asserted for any node mentioned by a
        signal whose outcome is a failure, or that is the target of a FAILED
        relationship.
        """
        if not node_ids:
            return
        since = (
            datetime.now(UTC) - timedelta(seconds=self._recent_window_seconds)
        ).isoformat()

        recent_rows = self._repo.execute_read(
            f"MATCH (s)-[:{REL_MENTIONS}]->(n) "
            "WHERE n.id IN $ids "
            "AND (s.timestamp IS NULL OR s.timestamp >= datetime($since)) "
            "RETURN DISTINCT n.id AS nid",
            {"ids": node_ids, "since": since},
        )
        for row in recent_rows:
            nid = row.get("nid")
            if nid:
                self._bridge.query(f"assertz(kg_recent({_atom(nid)}))")

        failed_rows = self._repo.execute_read(
            f"MATCH (s)-[:{REL_MENTIONS}]->(n) "
            "WHERE n.id IN $ids "
            "AND toLower(coalesce(s.outcome, '')) IN ['failure', 'failed', 'error'] "
            "RETURN DISTINCT n.id AS nid",
            {"ids": node_ids},
        )
        for row in failed_rows:
            nid = row.get("nid")
            if nid:
                self._bridge.query(f"assertz(kg_failed({_atom(nid)}))")

    def refine(self, subgraph: RankedSubgraph) -> dict[str, str]:
        """Return ``{node_id: tier}`` for every node in ``subgraph``.

        Empty subgraph => empty map. Every selected node gets exactly one tier;
        nodes the rules do not promote default to ``peripheral``. Projected
        facts are cleared before and after (REQ-106-064).
        """
        if not subgraph.nodes:
            return {}
        self._ensure_loaded()
        self._clear()
        try:
            self._project(subgraph)
            self._project_runtime_signals([n.node_id for n in subgraph.nodes])
            tiers: dict[str, str] = {n.node_id: TIER_PERIPHERAL for n in subgraph.nodes}
            # core beats related beats peripheral when a node proves several.
            result = self._bridge.query("kg_relevance_tier(N, T)")
            for sol in result.solutions:
                node_id, tier = sol.get("N"), sol.get("T")
                if node_id is None or tier is None:
                    continue
                if _tier_rank(tier) < _tier_rank(tiers.get(node_id, TIER_PERIPHERAL)):
                    tiers[node_id] = tier
            return tiers
        finally:
            self._clear()


def _tier_rank(tier: str) -> int:
    return {TIER_CORE: 0, TIER_RELATED: 1, TIER_PERIPHERAL: 2}.get(tier, 2)