orchestrata_ultimate.knowledge_graph.signal_extraction

Reference › orchestrata_ultimate › knowledge_graph › signal_extraction

Small helpers that pull file paths and code entity references out of runtime payloads. The Tier-H runtime signal writers use them to decide which File, class or function nodes a message, decision or tool call MENTIONS (see runtime_writer); the volk base class and the kernel's pipeline_kg import them. extract_file_paths walks strings, lists and dict values recursively with a regular expression for common source extensions, keeps only candidates that contain a /, and returns at most ten sorted, deduplicated paths. extract_entity_ids flattens the payload to text with _to_text and matches the relpath::qualname identity produced by the AST sync, returning at most ten in first-seen order. Extraction is best-effort and never raises. The module reads and writes nothing itself.

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/knowledge_graph/signal_extraction.py

"""Extract file paths and code entity references from runtime payloads.

Used by the Tier-H runtime signal writers to decide which ``File``/class/
function nodes a message, decision, or tool call ``MENTIONS``. Extraction is
best-effort and never raises.
"""

from __future__ import annotations

import re
from typing import Any

# Heuristic: what looks like a Python/JS/PL source file path.
_FILE_RE = re.compile(r"([\w./-]+\.(?:py|pl|js|ts|java|kt|rs|go|cpp|c|h|hpp|rb|sh))")
# Entity id pattern produced by AST sync: relpath::qualname.
_ENTITY_RE = re.compile(r"([\w./-]+::[\w.]+)")


def extract_file_paths(value: Any) -> list[str]:
    """Return candidate file paths found anywhere in ``value``.

    Walks strings, lists, and dict values recursively. Only paths that contain
    a ``/`` and a recognized source extension are returned, deduplicated, and
    capped at 10.
    """
    candidates: set[str] = set()
    _collect(
        str(value) if isinstance(value, str | int | float | bool) else value, candidates
    )
    return sorted(p for p in candidates if "/" in p)[:10]


def extract_entity_ids(value: Any) -> list[str]:
    """Return candidate canonical entity ids found anywhere in ``value``.

    Matches ``<relpath>::<qualname>`` patterns. Deduplicated, capped at 10.
    """
    text = _to_text(value)
    matches = _ENTITY_RE.findall(text)
    seen: set[str] = set()
    result: list[str] = []
    for m in matches:
        if m not in seen:
            seen.add(m)
            result.append(m)
    return result[:10]


def _to_text(value: Any) -> str:
    """Flatten a payload to a searchable text string."""
    if isinstance(value, str):
        return value
    if isinstance(value, dict):
        parts: list[str] = [s for k, v in value.items() for s in (str(k), _to_text(v))]
        return " ".join(parts)
    if isinstance(value, list | tuple | set):
        return " ".join(_to_text(v) for v in value)
    return str(value)


def _collect(value: Any, out: set[str]) -> None:
    """Recursively collect file path candidates into ``out``."""
    if isinstance(value, str):
        for match in _FILE_RE.findall(value):
            # Strip surrounding quotes/backticks that regex may have captured.
            match = match.strip("`\"'")
            if match:
                out.add(match)
    elif isinstance(value, dict):
        for k, v in value.items():
            _collect(str(k), out)
            _collect(v, out)
    elif isinstance(value, list | tuple | set):
        for item in value:
            _collect(item, out)