orchestrata_ultimate.knowledge_graph.context_pack.frame

Reference › orchestrata_ultimate › knowledge_graph › context_pack › frame

Deterministic request-frame analysis, the first step of the context pack pipeline (SPEC-106 FR-002). analyze_request turns free text or a task payload dictionary into a frozen RequestFrame holding a task type, the linked entities, a token budget and the requester, with no LLM or embedding call anywhere (ruling D11). The task type is classified by a table (_classify_task_type, reusing the SUBJECT_TASK_TYPE table and TaskType enum of context_injector, REQ-106-011) unless the caller passes one. Entity linking tokenises the text (_tokenize), fetches the canonical names of Module, Class, Function, Spec, Requirement and Volk nodes from the operational graph (orchkg on orchgraph) with _fetch_candidate_names, and matches case-, underscore- and dash-folded tokens (_fold, REQ-106-012). Each match becomes a LinkedEntity (token, label, node id, ambiguity flag); a name that folds to several nodes is marked ambiguous rather than guessed. The result feeds selector.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

LinkedEntity (dataclass)

Decorators: dataclass(frozen=True) · Line 83

One request token linked to canonical graph node(s).

RequestFrame (dataclass)

Decorators: dataclass(frozen=True) · Line 102

FR-002's output: {task_type, entities, budget, requester}.

Functions

Source

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

"""Deterministic request-frame analysis (SPEC-106 FR-002).

``analyze_request`` turns free text (or a task payload dict) into a
:class:`RequestFrame` — a task type, a set of linked graph entities, a token
budget, and a requester — with NO LLM/embedding calls anywhere (D11, the
ruling constraint of plans/decisions-2026-07-06.md: selection is
deterministic end-to-end).

Task-type classification reuses the existing deterministic table
(``SUBJECT_TASK_TYPE``) and enum (``TaskType``) from
``knowledge_graph.context_injector`` — SPEC-106 explicitly does not invent a
second task-type taxonomy.

Entity linking matches request tokens against canonical node names in the
live graph (Module/Class/Function/Spec/Requirement/Volk — the FR-001 code +
spec + runtime identity labels a request plausibly names by a human-readable
token; Session/Project/Intent/PipelineRun/Decision/Artifact/RequirementItem
are runtime bookkeeping labels no one "requests" by name). Matching is
case/underscore/dash-folded (REQ-106-012); ambiguous matches (one folded
name resolving to more than one graph node) are KEPT with the ambiguity
recorded on the entity, never dropped or guessed (REQ-106-013).
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.canonical_schema import (
    CANONICAL_MERGE_KEYS,
    LABEL_CLASS,
    LABEL_FUNCTION,
    LABEL_MODULE,
    LABEL_REQUIREMENT,
    LABEL_SPEC,
    LABEL_VOLK,
    parse_code_entity_id,
)
from orchestrata_ultimate.knowledge_graph.context_injector import (
    SUBJECT_TASK_TYPE,
    TaskType,
)

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph import SyncGraphRepository

# The canonical labels a request's free-text tokens are linked against. A
# narrower set than canonical_schema.CANONICAL_LABELS: Session/Project/Intent/
# PipelineRun/Decision/Artifact/RequirementItem are runtime bookkeeping nodes,
# not things a human/agent names by a short token in a request.
LINKABLE_LABELS: tuple[str, ...] = (
    LABEL_MODULE,
    LABEL_CLASS,
    LABEL_FUNCTION,
    LABEL_SPEC,
    LABEL_REQUIREMENT,
    LABEL_VOLK,
)

_CODE_LABELS = frozenset({LABEL_MODULE, LABEL_CLASS, LABEL_FUNCTION})

# Token shape: word chars plus the separators canonical ids/aliases actually
# use (dash for SPEC-/REQ- ids, colon for Volk aliases like "U:Cd", dot for
# dotted qualnames). Deliberately excludes '/' — code-entity ids embed a
# relpath with '/', which free text never reproduces verbatim; qualname
# matching (see _fetch_candidate_names) is the realistic path for those.
_TOKEN_RE = re.compile(r"[A-Za-z0-9_.:\-]+")


def _fold(text: str) -> str:
    """Case/underscore/dash-fold a name for comparison (REQ-106-012)."""
    return text.casefold().replace("-", "").replace("_", "")


def _tokenize(text: str) -> list[str]:
    """Split free text into candidate entity-linking tokens, in order."""
    return _TOKEN_RE.findall(text or "")


@dataclass(frozen=True)
class LinkedEntity:
    """One request token linked to canonical graph node(s).

    When ``ambiguous`` is False, exactly one node matched: ``label``/
    ``node_id`` name it directly. When True, more than one node shared the
    folded name; per REQ-106-013 the match is KEPT (not dropped, not
    guessed) — ``label``/``node_id`` are left empty and every match is
    recorded in ``candidates`` as ``(label, node_id)`` pairs so downstream
    selection can still use (and penalize) all of them.
    """

    token: str
    label: str
    node_id: str
    ambiguous: bool
    candidates: tuple[tuple[str, str], ...] = ()


@dataclass(frozen=True)
class RequestFrame:
    """FR-002's output: {task_type, entities, budget, requester}."""

    task_type: TaskType
    entities: tuple[LinkedEntity, ...] = field(default_factory=tuple)
    budget: int = 0
    requester: str = "agent"


def _classify_task_type(subject: str | None, text: str) -> TaskType:
    """Table-driven task-type classification — no model calls (REQ-106-011).

    Resolution order: explicit ``subject`` against ``SUBJECT_TASK_TYPE``,
    then that same table's keys as keywords in ``text``, then the six
    ``TaskType`` values themselves as keywords. Falls back to
    :data:`TaskType.BUILD` — the most general task — when nothing matches,
    which keeps this function total (every input produces a frame) while
    staying a pure, deterministic function of ``(subject, text)``.
    """
    if subject:
        mapped = SUBJECT_TASK_TYPE.get(subject)
        if mapped is not None:
            return mapped
    tokens = {_fold(t) for t in _tokenize(text)}
    for subj, task_type in SUBJECT_TASK_TYPE.items():
        if _fold(subj) in tokens:
            return task_type
    for task_type in TaskType:
        if _fold(task_type.value) in tokens:
            return task_type
    return TaskType.BUILD


def _split_payload(text_or_payload: str | dict[str, Any]) -> tuple[str | None, str]:
    """Normalize the ``text | task payload`` union into ``(subject, text)``."""
    if isinstance(text_or_payload, dict):
        subject = text_or_payload.get("subject")
        text = (
            text_or_payload.get("text")
            or text_or_payload.get("target")
            or text_or_payload.get("description")
            or ""
        )
        return subject, str(text)
    return None, str(text_or_payload)


def _fetch_candidate_names(
    repo: SyncGraphRepository,
) -> dict[str, list[tuple[str, str]]]:
    """Build ``{folded_name: [(label, node_id), ...]}`` from the live graph.

    One label at a time, ``ORDER BY`` the canonical key, so the resulting
    per-name candidate lists have a fixed, reproducible order given fixed
    graph content (REQ-106-014's "pure function of input + graph names").
    """
    candidates: dict[str, list[tuple[str, str]]] = {}
    for label in LINKABLE_LABELS:
        key = CANONICAL_MERGE_KEYS[label]
        result = repo.execute_read(
            f"MATCH (n:`{label}`) WHERE n.`{key}` IS NOT NULL "
            f"RETURN n.`{key}` AS node_id, n.name AS name, "
            "n.qualified_name AS qualified_name ORDER BY node_id"
        )
        for record in result:
            node_id = record["node_id"]
            names = {_fold(node_id)}
            if label in _CODE_LABELS:
                # PRIMARY: the node's own name/qualified_name properties —
                # id-format-agnostic. The ruling-2 migration moved code ids to
                # "{project}:{qualname}", which the legacy "::" parser below
                # rejects; linking silently returned EMPTY frames for every
                # code-entity request until this path was added (2026-07-21).
                qualname = record.get("qualified_name")
                if qualname:
                    names.add(_fold(qualname))
                    names.add(_fold(qualname.rpartition(".")[-1]))
                short_name = record.get("name")
                if short_name:
                    names.add(_fold(short_name))
                # FALLBACK: legacy "<relpath>::<qualname>" ids (property-less
                # nodes, e.g. pre-migration graphs and test seeds).
                try:
                    _, parsed_qualname = parse_code_entity_id(node_id)
                except ValueError:
                    pass
                else:
                    names.add(_fold(parsed_qualname))
                    # Also register the qualname's last dotted segment
                    # (e.g. "func_a" out of "ClassA.func_a") — a request
                    # naming just a method/function by its short name,
                    # without its owning class, is the common case.
                    names.add(_fold(parsed_qualname.rpartition(".")[-1]))
            for name in names:
                candidates.setdefault(name, []).append((label, node_id))
    return candidates


def _link_entities(text: str, repo: SyncGraphRepository) -> tuple[LinkedEntity, ...]:
    """Link request tokens to canonical graph node(s) (REQ-106-012/013)."""
    candidates_by_name = _fetch_candidate_names(repo)
    entities: list[LinkedEntity] = []
    seen: set[str] = set()
    for token in _tokenize(text):
        folded = _fold(token)
        if not folded or folded in seen:
            continue
        seen.add(folded)
        matches = candidates_by_name.get(folded)
        if not matches:
            continue
        unique_matches = list(dict.fromkeys(matches))
        if len(unique_matches) == 1:
            label, node_id = unique_matches[0]
            entities.append(
                LinkedEntity(token=token, label=label, node_id=node_id, ambiguous=False)
            )
        else:
            entities.append(
                LinkedEntity(
                    token=token,
                    label="",
                    node_id="",
                    ambiguous=True,
                    candidates=tuple(sorted(unique_matches)),
                )
            )
    return tuple(entities)


@traces(
    spec="106",
    reqs=[
        "REQ-106-010",
        "REQ-106-011",
        "REQ-106-012",
        "REQ-106-014",
        "REQ-106-013",
        "REQ-106-033",
        "REQ-106-034",
        "REQ-106-035",
    ],
)
def analyze_request(
    text_or_payload: str | dict[str, Any],
    *,
    repo: SyncGraphRepository,
    task_type: TaskType | None = None,
    budget: int,
    requester: str,
) -> RequestFrame:
    """Build a deterministic :class:`RequestFrame` (FR-002).

    Args:
        text_or_payload: Free text, or a task payload dict carrying any of
            ``subject`` (a CXP-style subject looked up in
            ``SUBJECT_TASK_TYPE``), ``text``/``target``/``description`` (the
            entity-linking text).
        repo: A connected ``SyncGraphRepository`` (entity linking is the
            only I/O this function performs; everything else is pure).
        task_type: Explicit override — when given, task-type classification
            is skipped entirely.
        budget: Token budget carried through to selection/packing.
        requester: ``"volk"`` | ``"agent"`` | ``"human"`` — who this frame
            is being built for (cosmetic, carried through unchanged).

    Returns:
        A :class:`RequestFrame`. Given the same input and the same graph
        state, this is a pure function (REQ-106-014): identical text yields
        an identical frame.
    """
    subject, text = _split_payload(text_or_payload)
    resolved_task_type = (
        task_type if task_type is not None else _classify_task_type(subject, text)
    )
    entities = _link_entities(text, repo)
    return RequestFrame(
        task_type=resolved_task_type,
        entities=entities,
        budget=budget,
        requester=requester,
    )