orchestrata_ultimate.knowledge_graph.temporal

Reference › orchestrata_ultimate › knowledge_graph › temporal

TemporalManager keeps a validity-interval history of node properties (Spec 008 REQ-008-002, -027 to -029, -083 to -086 and Spec 015 REQ-015-170 onward). Versions are held in a per-node list in process memory; each is a dictionary with a UUID, the properties, valid_from, valid_until and changed_by. version_node stamps the current time, closes the previous version by setting its valid_until, appends a new open-ended one, and, if a graph_client is supplied and connected, also calls update_node to write the properties and an updated_at stamp to the operational graph (orchkg on orchgraph). A persistence failure is logged as a warning and does not fail the call. get_state_at parses an ISO timestamp (naive values are taken as UTC), scans versions newest first and returns a Node built from the version whose interval contains it, or None. get_history, invalidate, get_version_count and clear complete it. The ITemporalManager contract is in interfaces.

At a glance

Class diagram

Dependency diagram

API

Classes

TemporalManager

Decorators: traces(spec='008', story='US-001', reqs=['REQ-008-002', 'REQ-008-029']), traces(spec='015', story='US-001', reqs=['REQ-015-170']) · Line 29

Manages temporal versioning of knowledge graph nodes.

Functions

Source

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

# mypy: ignore-errors
"""Temporal versioning for the knowledge graph.

Manages time-aware node states with valid_from/valid_until
semantics for point-in-time queries.
"""

from __future__ import annotations

import logging
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.types import ChangeEntry, ChangeType, Node

if TYPE_CHECKING:
    from orchestrata_ultimate.knowledge_graph.graph_client import GraphClient

logger = logging.getLogger(__name__)


def _utc_now() -> datetime:
    return datetime.now(UTC)


@traces(spec="008", story="US-001", reqs=["REQ-008-002", "REQ-008-029"])
@traces(spec="015", story="US-001", reqs=["REQ-015-170"])
class TemporalManager:
    """Manages temporal versioning of knowledge graph nodes.

    Implements ITemporalManager protocol.
    Each node version has valid_from and valid_until timestamps.
    """

    def __init__(self, graph_client: GraphClient | None = None) -> None:
        self._client = graph_client
        self._versions: dict[str, list[dict[str, Any]]] = {}

    @traces(
        spec="008",
        story="US-001",
        reqs=["REQ-008-027", "REQ-008-083", "REQ-008-084", "REQ-008-085"],
    )
    @traces(spec="015", story="US-001", reqs=["REQ-015-172"])
    async def version_node(
        self, node_id: str, properties: dict[str, Any], changed_by: str = ""
    ) -> str:
        """Create a new version of a node. Returns version ID."""
        now = _utc_now()

        # Close previous version
        versions = self._versions.get(node_id, [])
        if versions:
            versions[-1]["valid_until"] = now.isoformat()

        # Create new version
        import uuid

        version_id = str(uuid.uuid4())
        version = {
            "version_id": version_id,
            "node_id": node_id,
            "properties": dict(properties),
            "valid_from": now.isoformat(),
            "valid_until": None,
            "changed_by": changed_by,
        }
        versions.append(version)
        self._versions[node_id] = versions

        # Persist to the graph if available
        if self._client and self._client.connected:
            try:
                await self._client.update_node(
                    node_id,
                    {
                        "updated_at": now.isoformat(),
                        **properties,
                    },
                )
            except Exception as exc:
                logger.warning("Failed to persist version to the graph: %s", exc)

        return version_id

    @traces(
        spec="008", story="US-001", reqs=["REQ-008-028", "REQ-008-047", "REQ-008-086"]
    )
    @traces(spec="015", story="US-001", reqs=["REQ-015-173"])
    async def get_state_at(self, node_id: str, timestamp: str) -> Node | None:
        """Get the state of a node at a specific point in time."""
        ts = datetime.fromisoformat(timestamp)
        if ts.tzinfo is None:
            ts = ts.replace(tzinfo=UTC)
        versions = self._versions.get(node_id, [])

        for version in reversed(versions):
            valid_from = datetime.fromisoformat(version["valid_from"])
            if valid_from.tzinfo is None:
                valid_from = valid_from.replace(tzinfo=UTC)
            valid_until = version.get("valid_until")

            if valid_from <= ts:
                if valid_until is None:
                    return Node(
                        id=node_id,
                        name=version["properties"].get("name", ""),
                        properties=version["properties"],
                    )
                valid_until_dt = datetime.fromisoformat(valid_until)
                if valid_until_dt.tzinfo is None:
                    valid_until_dt = valid_until_dt.replace(tzinfo=UTC)
                if valid_until_dt > ts:
                    return Node(
                        id=node_id,
                        name=version["properties"].get("name", ""),
                        properties=version["properties"],
                    )
        return None

    @traces(
        spec="003",
        story="US-005",
        reqs=[
            "REQ-003-US-005-033",
            "REQ-003-US-006-009",
            "REQ-003-US-006-010",
            "REQ-003-US-006-012",
            "REQ-003-US-006-032",
        ],
    )
    @traces(
        spec="035", story="US-005", reqs=["REQ-035-US-004-008", "REQ-035-US-009-013"]
    )
    async def get_history(self, node_id: str, limit: int = 100) -> list[ChangeEntry]:
        """Get change history for a node."""
        versions = self._versions.get(node_id, [])
        entries = []

        for i, version in enumerate(versions[-limit:]):
            change_type = ChangeType.CREATED if i == 0 else ChangeType.MODIFIED
            entry = ChangeEntry(
                entity_id=node_id,
                entity_type="node",
                change_type=change_type,
                new_value=version["properties"],
                old_value=versions[i - 1]["properties"] if i > 0 else None,
                changed_by=version.get("changed_by", ""),
                changed_at=datetime.fromisoformat(version["valid_from"]),
            )
            entries.append(entry)

        return entries

    @traces(spec="008", story="US-001", reqs=["REQ-008-027", "REQ-008-084"])
    @traces(spec="015", story="US-001", reqs=["REQ-015-175"])
    async def invalidate(self, node_id: str) -> bool:
        """Mark the current version of a node as invalid."""
        versions = self._versions.get(node_id, [])
        if not versions:
            return False
        versions[-1]["valid_until"] = _utc_now().isoformat()
        return True

    def get_version_count(self, node_id: str) -> int:
        """Get the number of versions for a node."""
        return len(self._versions.get(node_id, []))

    def clear(self) -> None:
        """Clear all version data."""
        self._versions.clear()