orchestrata_ultimate.knowledge_graph.context_pack.packer

Reference › orchestrata_ultimate › knowledge_graph › context_pack › packer

Budgeted packing and the editable pack format, the last step of the context pack pipeline (SPEC-106 FR-004). pack takes the ranked nodes from selector best first and includes them greedily under an explicit token budget that is never exceeded (REQ-106-030); tokens are counted with the shared TokenCounterProtocol implementations from the memory package (REQ-106-031). The result is a frozen ContextPack with the request frame, a snapshot_id, the budget and the amount spent, and a list of PackSection entries each carrying node id, label, selecting template, rank score, token cost and body (REQ-106-033).

A pack renders as a markdown file: YAML frontmatter (frame, snapshot id, budget and spent) followed by one section per node with full provenance. The snapshot id is a hash over the selected nodes' ids and rendered content only, with no wall clock, so the same frame against the same graph state gives a byte-identical body (REQ-106-034). render produces the text and parse recovers a ParsedPack of ParsedSection records from it. Packs are meant to be hand-trimmed by deleting whole sections and re-injected without regeneration, and validate_pack checks that a trimmed pack is still structurally complete (REQ-106-035). context_pack_cache caches rendered packs. The packer reads and writes no database.

At a glance

Class diagram

Dependency diagram

API

Classes

PackSection (dataclass)

Decorators: dataclass(frozen=True) · Line 48

One rendered pack section with full provenance (REQ-106-033).

ContextPack (dataclass)

Decorators: dataclass(frozen=True) · Line 60

FR-004's output — a budgeted, provenance-carrying, byte-stable pack.

ParsedSection (dataclass)

Decorators: traces(spec='106', reqs=['REQ-106-033', 'REQ-106-035']), dataclass(frozen=True) · Line 73

A section recovered from a rendered pack's text by :func:parse.

ParsedPack (dataclass)

Decorators: traces(spec='106', reqs=['REQ-106-032', 'REQ-106-035']), dataclass(frozen=True) · Line 86

A rendered pack's text recovered back into frontmatter + sections.

Functions

Source

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

"""Budgeted packing + the editable pack format (SPEC-106 FR-004).

``pack()`` greedily includes ranked nodes (best-first, as produced by
:func:`selector.select`) under an explicit token budget that is never
exceeded (REQ-106-030), counted via the shared
``memory.token_counter.TokenCounterProtocol`` implementations (REQ-106-031).

A pack is a markdown FILE: YAML frontmatter (request frame, graph snapshot
id, budget/spent) followed by one section per selected node, each carrying
full provenance — node id, selecting template, rank score, token cost
(REQ-106-032/033). The snapshot id is a hash over the *selected* nodes' ids
and rendered content only — no wall-clock anywhere — so the same frame
against the same graph state renders a byte-identical pack body
(REQ-106-034).

Packs are meant to be hand-trimmed (whole sections deleted) and re-injected
without regeneration; :func:`validate_pack` re-validates that a (possibly
trimmed) rendered pack is still structurally complete (REQ-106-035).
"""

from __future__ import annotations

import hashlib
import json
import re
from dataclasses import dataclass
from typing import Any

import yaml

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import TokenCounterProtocol
from orchestrata_ultimate.knowledge_graph.context_pack.frame import RequestFrame
from orchestrata_ultimate.knowledge_graph.context_pack.selector import (
    RankedNode,
    RankedSubgraph,
)
from orchestrata_ultimate.memory.token_counter import get_token_counter

_HEADING_RE = re.compile(r"^## (?P<label>\S+) (?P<node_id>.+)$")
_PROVENANCE_RE = re.compile(
    r"^<!-- provenance: node_id=(?P<node_id>\S+) template=(?P<template>\S+) "
    r"rank=(?P<rank>-?[\d.]+) tokens=(?P<tokens>\d+) -->$"
)


@dataclass(frozen=True)
class PackSection:
    """One rendered pack section with full provenance (REQ-106-033)."""

    node_id: str
    label: str
    template: str
    rank_score: float
    token_cost: int
    body: str  # rendered "## Label id\n<property bullets>" content


@dataclass(frozen=True)
class ContextPack:
    """FR-004's output — a budgeted, provenance-carrying, byte-stable pack."""

    frame: RequestFrame
    snapshot_id: str
    budget: int
    spent: int
    sections: tuple[PackSection, ...] = ()
    reason: str | None = None


@traces(spec="106", reqs=["REQ-106-033", "REQ-106-035"])
@dataclass(frozen=True)
class ParsedSection:
    """A section recovered from a rendered pack's text by :func:`parse`."""

    node_id: str
    label: str
    template: str
    rank_score: float
    token_cost: int
    body: str


@traces(spec="106", reqs=["REQ-106-032", "REQ-106-035"])
@dataclass(frozen=True)
class ParsedPack:
    """A rendered pack's text recovered back into frontmatter + sections."""

    frontmatter: dict[str, Any]
    sections: tuple[ParsedSection, ...] = ()


def _properties_lines(properties: dict[str, Any]) -> list[str]:
    """Render a node's properties as sorted, deterministic bullet lines."""
    lines = []
    for key in sorted(properties):
        value = properties[key]
        if value is None:
            continue
        lines.append(f"- **{key}**: `{value}`")
    return lines


def _render_section_content(node: RankedNode) -> str:
    """Render one node's substantive (heading + properties) markdown block."""
    heading = f"## {node.label} {node.node_id}"
    body = "\n".join(_properties_lines(node.properties)) or "(no properties)"
    return f"{heading}\n{body}"


def _snapshot_id(sections: list[PackSection]) -> str:
    """Deterministic hash over the selected nodes' ids + rendered content.

    No wall-clock, no random ids — the same set of sections in the same
    order always hashes to the same value (REQ-106-034).
    """
    payload = [
        {
            "id": s.node_id,
            "template": s.template,
            "rank": round(s.rank_score, 6),
            "body": s.body,
        }
        for s in sections
    ]
    blob = json.dumps(payload, sort_keys=True, ensure_ascii=True)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16]


@traces(
    spec="106",
    reqs=["REQ-106-030", "REQ-106-031", "REQ-106-033", "REQ-106-034", "REQ-106-035"],
)
def pack(
    subgraph: RankedSubgraph,
    frame: RequestFrame,
    *,
    counter: TokenCounterProtocol | None = None,
) -> ContextPack:
    """Greedily pack ``subgraph``'s ranked nodes under ``frame.budget`` (FR-004).

    Nodes are tried in rank order (best first — ``subgraph.nodes`` is
    already sorted by :func:`selector.select`); a node is included iff its
    rendered content fits the REMAINING budget. Lower-ranked nodes that
    still fit after a higher-ranked one was skipped are still tried
    (classic greedy-by-value packing) — the budget is never exceeded
    (REQ-106-030).

    Args:
        subgraph: A :class:`RankedSubgraph` from :func:`selector.select`.
        frame: The :class:`RequestFrame` this subgraph was selected for —
            its ``budget`` governs packing.
        counter: Token counter to use; defaults to
            ``memory.token_counter.get_token_counter()`` (word-count
            fallback — REQ-106-031's TokenCounterProtocol contract, no
            provider needed for a generic pack).

    Returns:
        A :class:`ContextPack` whose ``spent`` never exceeds ``budget``.
    """
    tc = counter if counter is not None else get_token_counter()
    spent = 0
    sections: list[PackSection] = []
    for node in subgraph.nodes:
        content = _render_section_content(node)
        cost = tc.count(content)
        if spent + cost > frame.budget:
            continue
        spent += cost
        sections.append(
            PackSection(
                node_id=node.node_id,
                label=node.label,
                template=node.template,
                rank_score=node.rank_score,
                token_cost=cost,
                body=content,
            )
        )
    return ContextPack(
        frame=frame,
        snapshot_id=_snapshot_id(sections),
        budget=frame.budget,
        spent=spent,
        sections=tuple(sections),
    )


@traces(spec="106", reqs=["REQ-106-043"])
def empty_pack(frame: RequestFrame, reason: str) -> ContextPack:
    """Build an explicit empty pack carrying a degradation reason.

    Used when the KG is unavailable — never a silent partial: the rendered
    frontmatter always says why there are zero sections (REQ-106-043).
    """
    return ContextPack(
        frame=frame,
        snapshot_id="",
        budget=frame.budget,
        spent=0,
        sections=(),
        reason=reason,
    )


def _frame_frontmatter(frame: RequestFrame) -> dict[str, Any]:
    return {
        "task_type": frame.task_type.value,
        "budget": frame.budget,
        "requester": frame.requester,
        "entities": [
            {
                "token": e.token,
                "label": e.label,
                "node_id": e.node_id,
                "ambiguous": e.ambiguous,
                "candidates": [list(c) for c in e.candidates],
            }
            for e in frame.entities
        ],
    }


@traces(
    spec="106",
    reqs=["REQ-106-032", "REQ-106-033", "REQ-106-034", "REQ-106-035", "REQ-106-043"],
)
def render(pack: ContextPack) -> str:
    """Render a :class:`ContextPack` to its markdown-with-frontmatter file body.

    Byte-stable (REQ-106-034): every value comes from ``pack`` itself
    (already computed deterministically by :func:`select`/`pack`) — no
    wall-clock, no random ids, no locale-dependent formatting.
    """
    frontmatter: dict[str, Any] = {
        "frame": _frame_frontmatter(pack.frame),
        "snapshot_id": pack.snapshot_id,
        "budget": pack.budget,
        "spent": pack.spent,
        "sections": len(pack.sections),
    }
    if pack.reason is not None:
        frontmatter["reason"] = pack.reason
    yaml_text = yaml.safe_dump(frontmatter, sort_keys=True, default_flow_style=False)
    parts = [f"---\n{yaml_text.rstrip(chr(10))}\n---"]
    for section in pack.sections:
        provenance = (
            f"<!-- provenance: node_id={section.node_id} template={section.template} "
            f"rank={section.rank_score:.6f} tokens={section.token_cost} -->"
        )
        parts.append(f"{section.body}\n{provenance}")
    return "\n\n".join(parts) + "\n"


@traces(spec="106", reqs=["REQ-106-035", "REQ-106-033", "REQ-106-043"])
def parse(text: str) -> ParsedPack:
    """Recover a rendered pack's frontmatter + sections from its text.

    Tolerant of hand-trimming (whole sections deleted): every remaining
    section block is parsed independently.
    """
    blocks = text.strip("\n").split("\n\n")
    if not blocks:
        return ParsedPack(frontmatter={}, sections=())

    frontmatter: dict[str, Any] = {}
    fm_match = re.match(r"^---\n(?P<yaml>.*)\n---$", blocks[0], re.DOTALL)
    if fm_match:
        frontmatter = yaml.safe_load(fm_match.group("yaml")) or {}

    sections: list[ParsedSection] = []
    for block in blocks[1:]:
        block = block.strip("\n")
        if not block:
            continue
        block_lines = block.split("\n")
        heading_match = _HEADING_RE.match(block_lines[0])
        provenance_match = _PROVENANCE_RE.match(block_lines[-1])
        if not heading_match or not provenance_match:
            continue
        body_lines = block_lines[:-1]
        sections.append(
            ParsedSection(
                node_id=provenance_match.group("node_id"),
                label=heading_match.group("label"),
                template=provenance_match.group("template"),
                rank_score=float(provenance_match.group("rank")),
                token_cost=int(provenance_match.group("tokens")),
                body="\n".join(body_lines),
            )
        )
    return ParsedPack(frontmatter=frontmatter, sections=tuple(sections))


@traces(spec="106", reqs=["REQ-106-035", "REQ-106-043"])
def validate_pack(text: str) -> bool:
    """Re-validate a (possibly hand-trimmed) pack as still injectable.

    Deleting whole sections keeps this True — each remaining section is
    independently self-describing (full provenance + non-empty body) and
    the frontmatter is untouched. Structural corruption (a missing/mangled
    provenance line, an unparsable frontmatter, an empty body) fails
    validation (REQ-106-035).
    """
    if not text.strip():
        return False
    parsed = parse(text)
    required_keys = {"frame", "snapshot_id", "budget", "spent"}
    if not required_keys.issubset(parsed.frontmatter):
        return False
    blocks = text.strip("\n").split("\n\n")
    non_frontmatter_blocks = [b for b in blocks[1:] if b.strip()]
    if len(non_frontmatter_blocks) != len(parsed.sections):
        # Some non-empty block failed to parse into a section — corrupted,
        # not merely trimmed.
        return False
    return all(section.body.strip() for section in parsed.sections)