orchestrata_ultimate.knowledge_graph.context_pack.cache

Reference › orchestrata_ultimate › knowledge_graph › context_pack › cache

ContextPackCache is an in-memory cache of rendered context pack markdown (SPEC-106, REQ-106-067). Because a rendered pack is deterministic for a given request frame, selected subgraph and task type, and the packer's snapshot_id captures the selected nodes' ids and content, repeated calls inside one pipeline run can skip the select, pack and render steps. The cache is thread-safe, process-local and bounded by a time to live (300 seconds by default), so it never survives a run and graph updates become visible without explicit invalidation. get returns the cached markdown for a ContextPack, keyed with the task type and an optional volk alias; put stores it; invalidate clears everything; size counts live entries, ignoring expired ones. context_injector and the kernel pipeline use it. It performs no I/O.

At a glance

Class diagram

Dependency diagram

API

Classes

ContextPackCache

Decorators: traces(spec='106', reqs=['REQ-106-067']) · Line 31

Thread-safe cache for rendered context packs.

Source

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

"""In-memory render cache for context packs.

A rendered context pack is deterministic for a given ``RequestFrame``,
selected subgraph, and task type (the packer's ``snapshot_id`` captures the
selected nodes' ids and rendered content). Caching the rendered markdown
avoids re-running the selector/packer/render pipeline for repeated calls
within the same pipeline run.

The cache is intentionally process-local and TTL-bounded: it never survives
across runs and it invalidates entries older than a configurable TTL so graph
updates are eventually visible even without explicit invalidation.
"""

from __future__ import annotations

import logging
import threading
import time
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.context_pack.packer import ContextPack

if TYPE_CHECKING:
    from orchestrata_ultimate.knowledge_graph.context_injector import TaskType

_logger = logging.getLogger(__name__)


@traces(spec="106", reqs=["REQ-106-067"])
class ContextPackCache:
    """Thread-safe cache for rendered context packs.

    Keys are ``(snapshot_id, task_type, volk_alias)``. The TTL defaults to
    5 minutes — long enough to benefit from repeated calls in a single
    pipeline run, short enough that stale graph state does not linger.

    Args:
        ttl_seconds: Time-to-live for cache entries. ``None`` disables TTL.
    """

    def __init__(self, ttl_seconds: float = 300.0) -> None:
        self._ttl_seconds = ttl_seconds
        self._lock = threading.Lock()
        self._store: dict[tuple[str, str, str], tuple[str, float]] = {}

    def _now(self) -> float:
        return time.monotonic()

    def get(
        self,
        pack: ContextPack,
        *,
        task_type: TaskType,
        volk_alias: str = "",
    ) -> str | None:
        """Return the cached rendered markdown for ``pack`` if available.

        Args:
            pack: The packed context (its ``snapshot_id`` is the cache key).
            task_type: Task type used for the pack.
            volk_alias: Volk alias used for the pack.

        Returns:
            Rendered markdown string, or ``None`` when not cached or expired.
        """
        key = (pack.snapshot_id, task_type.value, volk_alias)
        with self._lock:
            entry = self._store.get(key)
            if entry is None:
                return None
            rendered, inserted = entry
            if (
                self._ttl_seconds is not None
                and self._now() - inserted > self._ttl_seconds
            ):
                self._store.pop(key, None)
                return None
            return rendered

    def put(
        self,
        rendered: str,
        pack: ContextPack,
        *,
        task_type: TaskType,
        volk_alias: str = "",
    ) -> None:
        """Cache a rendered markdown for ``pack``.

        Empty rendered strings or empty snapshot ids are not cached.
        """
        if not rendered or not pack.snapshot_id:
            return
        key = (pack.snapshot_id, task_type.value, volk_alias)
        with self._lock:
            self._store[key] = (rendered, self._now())

    def invalidate(self) -> None:
        """Clear all cached entries."""
        with self._lock:
            self._store.clear()

    def size(self) -> int:
        """Return the number of live cached entries (expired entries are
        lazily evicted on access, so this may overcount slightly)."""
        with self._lock:
            return len(self._store)