orchestrata_ultimate.orchestrator.intent_history

Reference › orchestrata_ultimate › orchestrator › intent_history

The searchable record of every intent the broker has handled. intent_broker is its only importer and takes it as an optional collaborator. Storage is dual: Redis through RedisStateManager for fast lookup, and an optional graph audit sink from graph_audit for the durable trail.

IntentHistory.record serializes an Intent to JSON, stores it as a hash with an expiry, and adds its id to sorted-set indexes by action, requester and status, plus a recent-intents index scored by time, all in one Redis pipeline. If a graph is wired, _persist_to_graph then logs an INTENT_RECORDED AuditEvent; a failure there is logged as a warning and does not fail the record. Reading back: get fetches one entry, search filters with an IntentHistoryFilter (matches tests one entry), paginate returns a Page (with has_next and total_pages), get_recent returns the latest, and replay returns an entry stamped with a replay time. export_json and export_csv dump filtered results, get_statistics returns a HistoryStats, count reports the size, and prune deletes entries older than a threshold (30 days by default) and returns how many it removed.

At a glance

Class diagram

Dependency diagram

API

Classes

HistoryStats (dataclass)

Decorators: traces(spec='003', story='US-009', reqs=['REQ-003-US-009-008']), dataclass · Line 42

Aggregated intent history statistics.

IntentHistoryFilter (dataclass)

Decorators: traces(spec='003', story='US-006', reqs=['REQ-003-US-006-003']), dataclass · Line 54

Filter criteria for history queries.

Page (dataclass)

Decorators: traces(spec='003', story='US-006', reqs=['REQ-003-US-006-007', 'REQ-003-US-006-018']), dataclass · Line 88

Paginated result set.

IntentHistory

Decorators: traces(spec='003', story='US-006', reqs=['REQ-003-US-006-001', 'REQ-003-US-006-011', 'REQ-003-US-006-012']) · Line 118

Dual-storage intent history manager.

Source

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

# mypy: ignore-errors
"""Intent history with dual Redis+graph storage.

Records intent lifecycle events, supports search, pagination,
replay, export, and statistics.
"""

import csv
import io
import json
import logging
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any, TypeVar

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.infra.redis_state import RedisStateManager
from orchestrata_ultimate.orchestrator.intent_types import (
    Intent,
    IntentAction,
    IntentStatus,
    IntentTarget,
)

logger = logging.getLogger(__name__)


T = TypeVar("T")

# Redis key prefixes
HISTORY_KEY = "orchestrata:intent:history"
HISTORY_INDEX_ACTION = "orchestrata:intent:history:index:action"
HISTORY_INDEX_REQUESTER = "orchestrata:intent:history:index:requester"
HISTORY_INDEX_STATUS = "orchestrata:intent:history:index:status"
HISTORY_RECENT = "orchestrata:intent:history:recent"
HISTORY_TTL = 86400 * 7  # 7 days


@traces(spec="003", story="US-009", reqs=["REQ-003-US-009-008"])
@dataclass
class HistoryStats:
    """Aggregated intent history statistics."""

    total_intents: int = 0
    by_action: dict[str, int] = field(default_factory=dict)
    by_status: dict[str, int] = field(default_factory=dict)
    by_target: dict[str, int] = field(default_factory=dict)
    avg_processing_ms: float = 0.0


@traces(spec="003", story="US-006", reqs=["REQ-003-US-006-003"])
@dataclass
class IntentHistoryFilter:
    """Filter criteria for history queries."""

    action: IntentAction | None = None
    target: IntentTarget | None = None
    status: IntentStatus | None = None
    requester: str | None = None
    from_date: datetime | None = None
    to_date: datetime | None = None

    @traces(spec="003", story="US-006", reqs=["REQ-003-US-006-003"])
    def matches(self, data: dict[str, Any]) -> bool:
        """Check if a history entry matches this filter."""
        if self.action and data.get("action") != self.action.value:
            return False
        if self.target and data.get("target") != self.target.value:
            return False
        if self.status and data.get("status") != self.status.value:
            return False
        if self.requester and data.get("requester") != self.requester:
            return False
        if self.from_date:
            created = data.get("created_at", "")
            if created and created < self.from_date.isoformat():
                return False
        if self.to_date:
            created = data.get("created_at", "")
            if created and created > self.to_date.isoformat():
                return False
        return True


@traces(spec="003", story="US-006", reqs=["REQ-003-US-006-007", "REQ-003-US-006-018"])
@dataclass
class Page[T]:
    """Paginated result set."""

    items: list[T]
    page: int
    page_size: int
    total: int

    @property
    @traces(
        spec="003", story="US-006", reqs=["REQ-003-US-006-007", "REQ-003-US-006-018"]
    )
    def has_next(self) -> bool:
        return (self.page * self.page_size) < self.total

    @property
    @traces(
        spec="003", story="US-006", reqs=["REQ-003-US-006-007", "REQ-003-US-006-018"]
    )
    def total_pages(self) -> int:
        if self.page_size <= 0:
            return 0
        return (self.total + self.page_size - 1) // self.page_size


@traces(
    spec="003",
    story="US-006",
    reqs=["REQ-003-US-006-001", "REQ-003-US-006-011", "REQ-003-US-006-012"],
)
class IntentHistory:
    """Dual-storage intent history manager.

    Uses Redis for fast recent queries and optional graph for
    long-term archival and graph-based analysis.
    """

    def __init__(
        self,
        redis: RedisStateManager,
        graph: Any = None,
    ) -> None:
        self._redis = redis
        self._graph = graph

    @property
    def client(self) -> Any:
        return self._redis.client

    @traces(
        spec="003",
        story="US-006",
        reqs=[
            "REQ-003-US-006-001",
            "REQ-003-US-006-010",
            "REQ-003-US-006-013",
            "REQ-003-US-006-021",
            "REQ-003-US-006-026",
            "REQ-003-US-006-032",
        ],
    )
    async def record(self, intent: Intent) -> None:
        """Record an intent to history."""
        data = intent.to_dict()
        data["recorded_at"] = datetime.now(UTC).isoformat()
        json_data = json.dumps(data)
        timestamp = time.time()

        pipe = self.client.pipeline()
        # Store the intent data
        pipe.hset(f"{HISTORY_KEY}:{intent.id}", mapping={"data": json_data})
        pipe.expire(f"{HISTORY_KEY}:{intent.id}", HISTORY_TTL)

        # Index by action
        pipe.zadd(
            f"{HISTORY_INDEX_ACTION}:{intent.action.value}",
            {intent.id: timestamp},
        )
        # Index by requester
        if intent.requester:
            pipe.zadd(
                f"{HISTORY_INDEX_REQUESTER}:{intent.requester}",
                {intent.id: timestamp},
            )
        # Index by status
        pipe.zadd(
            f"{HISTORY_INDEX_STATUS}:{intent.status.value}",
            {intent.id: timestamp},
        )
        # Recent intents (sliding window)
        pipe.zadd(HISTORY_RECENT, {intent.id: timestamp})

        pipe.execute()

        # Best-effort graph persistence
        if self._graph is not None:
            try:
                await self._persist_to_graph(intent)
            except Exception as exc:
                logger.warning(
                    "Graph persistence failed for intent %s: %s", intent.id, exc
                )

    @traces(spec="003", story="US-006", reqs=["REQ-003-US-006-025"])
    async def _persist_to_graph(self, intent: Intent) -> None:
        """Persist intent to graph."""
        if self._graph is None:
            return
        # Build a REAL AuditEvent. `log_event` reads `event_type.value`,
        # `timestamp.isoformat()`, `session_id`, `volk_id`, and `outcome.value`,
        # so the previous bare `type("AuditEvent", ...)` shim (event_type=str,
        # timestamp=str, missing fields) raised AttributeError — which record()'s
        # best-effort except swallowed, silently dropping every intent's
        # graph persistence.
        from orchestrata_ultimate.orchestrator.graph_audit import (
            AuditEvent,
            AuditEventType,
        )

        await self._graph.log_event(
            AuditEvent(
                event_type=AuditEventType.INTENT_RECORDED,
                session_id=str(intent.session_id or ""),
                volk_id=str(intent.requester or ""),
                timestamp=datetime.now(UTC),
                data=intent.to_dict(),
                correlation_id=intent.correlation_id,
            )
        )

    @traces(spec="003", story="US-006", reqs=["REQ-003-US-006-015"])
    async def get(self, intent_id: str) -> dict[str, Any] | None:
        """Retrieve a single intent from history."""
        raw = self.client.hget(f"{HISTORY_KEY}:{intent_id}", "data")
        if raw is None:
            return None
        if isinstance(raw, bytes):
            raw = raw.decode()
        return json.loads(raw)

    @traces(
        spec="003",
        story="US-006",
        reqs=[
            "REQ-003-US-006-002",
            "REQ-003-US-006-006",
            "REQ-003-US-006-014",
            "REQ-003-US-006-017",
            "REQ-003-US-006-027",
            "REQ-003-US-006-028",
            "REQ-003-US-006-029",
            "REQ-003-US-006-035",
        ],
    )
    async def search(
        self,
        filters: IntentHistoryFilter | None = None,
        limit: int = 100,
    ) -> list[dict[str, Any]]:
        """Search history with optional filters."""
        # Determine which index to scan
        index_key = HISTORY_RECENT
        if filters:
            if filters.action:
                index_key = f"{HISTORY_INDEX_ACTION}:{filters.action.value}"
            elif filters.requester:
                index_key = f"{HISTORY_INDEX_REQUESTER}:{filters.requester}"
            elif filters.status:
                index_key = f"{HISTORY_INDEX_STATUS}:{filters.status.value}"

        # Get IDs from the index (most recent first)
        intent_ids = self.client.zrevrange(index_key, 0, limit * 2 - 1)

        results = []
        for raw_id in intent_ids:
            intent_id = raw_id.decode() if isinstance(raw_id, bytes) else raw_id
            entry = await self.get(intent_id)
            if entry is None:
                continue
            if filters and not filters.matches(entry):
                continue
            results.append(entry)
            if len(results) >= limit:
                break

        return results

    @traces(
        spec="003",
        story="US-006",
        reqs=["REQ-003-US-006-007", "REQ-003-US-006-018", "REQ-003-US-006-033"],
    )
    async def paginate(
        self,
        page: int = 1,
        page_size: int = 20,
        filters: IntentHistoryFilter | None = None,
    ) -> Page:
        """Paginated history query."""
        all_results = await self.search(
            filters=filters, limit=page_size * page + page_size
        )
        total = len(all_results)
        start = (page - 1) * page_size
        end = start + page_size
        items = all_results[start:end]

        return Page(
            items=items,
            page=page,
            page_size=page_size,
            total=total,
        )

    @traces(
        spec="003",
        story="US-006",
        reqs=[
            "REQ-003-US-006-004",
            "REQ-003-US-006-005",
            "REQ-003-US-006-016",
            "REQ-003-US-006-030",
            "REQ-003-US-006-031",
        ],
    )
    async def replay(self, intent_id: str) -> dict[str, Any] | None:
        """Retrieve intent data for replay purposes."""
        entry = await self.get(intent_id)
        if entry is not None:
            entry["replayed_at"] = datetime.now(UTC).isoformat()
        return entry

    @traces(
        spec="003",
        story="US-006",
        reqs=[
            "REQ-003-US-006-008",
            "REQ-003-US-006-019",
            "REQ-003-US-006-025",
            "REQ-003-US-006-034",
            "REQ-003-US-009-023",
        ],
    )
    async def export_json(
        self,
        filters: IntentHistoryFilter | None = None,
    ) -> str:
        """Export history as JSON string."""
        results = await self.search(filters=filters, limit=10000)
        return json.dumps(results, indent=2, default=str)

    @traces(
        spec="003", story="US-006", reqs=["REQ-003-US-006-008", "REQ-003-US-006-019"]
    )
    async def export_csv(
        self,
        filters: IntentHistoryFilter | None = None,
    ) -> str:
        """Export history as CSV string."""
        results = await self.search(filters=filters, limit=10000)
        if not results:
            return ""

        output = io.StringIO()
        fieldnames = ["id", "action", "target", "requester", "status", "created_at"]
        writer = csv.DictWriter(output, fieldnames=fieldnames, extrasaction="ignore")
        writer.writeheader()
        for entry in results:
            writer.writerow(entry)
        return output.getvalue()

    @traces(
        spec="003",
        story="US-009",
        reqs=["REQ-003-US-009-008", "REQ-003-US-009-021", "REQ-003-US-009-034"],
    )
    async def get_statistics(self) -> HistoryStats:
        """Compute aggregate statistics."""
        stats = HistoryStats()

        # Count by action
        for action in IntentAction:
            key = f"{HISTORY_INDEX_ACTION}:{action.value}"
            count = self.client.zcard(key)
            if count > 0:
                stats.by_action[action.value] = count
                stats.total_intents += count

        # Count by status
        for status in IntentStatus:
            key = f"{HISTORY_INDEX_STATUS}:{status.value}"
            count = self.client.zcard(key)
            if count > 0:
                stats.by_status[status.value] = count

        return stats

    @traces(
        spec="003",
        story="US-006",
        reqs=["REQ-003-US-006-009", "REQ-003-US-006-020", "REQ-003-US-006-023"],
    )
    async def prune(self, older_than_seconds: int = 86400 * 30) -> int:
        """Remove entries older than the given threshold."""
        cutoff = time.time() - older_than_seconds
        removed = 0

        # Get old IDs from the recent index
        old_ids = self.client.zrangebyscore(HISTORY_RECENT, 0, cutoff)

        pipe = self.client.pipeline()
        for raw_id in old_ids:
            intent_id = raw_id.decode() if isinstance(raw_id, bytes) else raw_id
            pipe.delete(f"{HISTORY_KEY}:{intent_id}")
            removed += 1

        # Clean the recent index
        pipe.zremrangebyscore(HISTORY_RECENT, 0, cutoff)
        pipe.execute()

        return removed

    @traces(spec="003", story="US-009", reqs=["REQ-003-US-009-008"])
    async def count(self) -> int:
        """Total number of recorded intents."""
        return self.client.zcard(HISTORY_RECENT)

    async def get_recent(self, limit: int = 10) -> list[dict[str, Any]]:
        """Get most recent intents."""
        intent_ids = self.client.zrevrange(HISTORY_RECENT, 0, limit - 1)
        results = []
        for raw_id in intent_ids:
            intent_id = raw_id.decode() if isinstance(raw_id, bytes) else raw_id
            entry = await self.get(intent_id)
            if entry:
                results.append(entry)
        return results