orchestrata_ultimate.core.protocols.memory

Reference › orchestrata_ultimate › core › protocols › memory

The memory-engine types of SPEC-010, used by the gateway server's memory wiring and re-exported by interfaces.memory. Two enums set policy: MemoryClassification (private, shared, system) and SummarizationStrategy (relevance, recency, importance). MemoryScope is a dataclass that narrows a search by session, task, volk, project, portfolio and date range. MemoryEntry is a Pydantic model with key, content, optional summary, tags, metadata, classification, embedding, timestamps, volk_id, session and task ids, a relevance score and a token count; volk_id is the canonical identity field and agent_id remains as a deprecated read-only alias. MemorySummary records a produced summary and the keys it came from.

TokenCounterProtocol has count and truncate. MemoryEngineProtocol is the engine contract: store, retrieve, search by query and scope, summarize a context under a token budget and strategy, update, delete and migrate_to_cold for moving a session's entries to cold storage. It builds on base.

At a glance

Class diagram

Dependency diagram

API

Classes

MemoryClassification (enum)

Bases: StrEnum · Decorators: traces(spec='010', story='US-006', reqs=['REQ-010-049']) · Line 29

Classification level for memory entries.

SummarizationStrategy (enum)

Bases: StrEnum · Line 37

Strategy for memory summarization.

MemoryScope (dataclass)

Decorators: dataclass, traces(spec='010', story='US-004', reqs=['REQ-010-005']) · Line 47

Scope filter for memory queries.

MemoryEntry (pydantic)

Bases: BaseModel · Decorators: traces(spec='010', story='US-001', reqs=['REQ-010-004']) · Line 65

A memory entry stored by an Agent.

MemorySummary (dataclass)

Decorators: dataclass, traces(spec='010', story='US-005', reqs=['REQ-010-023']) · Line 116

Result of memory summarization.

TokenCounterProtocol (protocol)

Bases: Protocol · Decorators: traces(spec='010', story='US-003', reqs=['REQ-010-041']) · Line 133

Provider-aware token counting abstraction.

MemoryEngineProtocol (protocol)

Bases: ComponentProtocol, Protocol · Decorators: traces(spec='010', story='US-001', reqs=['REQ-010-001']) · Line 153

Unified agent memory abstraction.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/protocols/memory.py

"""
Memory engine protocol types for Orchestrata Ultimate (SPEC-010).

Depends on :mod:`orchestrata_ultimate.core.protocols.base` for the
``ComponentProtocol`` base. ``MemoryEntry`` is a Pydantic model. This module is
the single source of truth: :mod:`orchestrata_ultimate.core.interfaces.memory`
re-exports these types as thin aliases (the interfaces->protocols twin-collapse).
"""

import json
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import StrEnum
from typing import Any, Protocol

from pydantic import BaseModel, ConfigDict
from pydantic import Field as PydanticField

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols.base import ComponentProtocol

# ============================================================================
# Memory Engine Types (SPEC-010)
# ============================================================================


@traces(spec="010", story="US-006", reqs=["REQ-010-049"])
class MemoryClassification(StrEnum):
    """Classification level for memory entries."""

    PRIVATE = "private"
    SHARED = "shared"
    SYSTEM = "system"


class SummarizationStrategy(StrEnum):
    """Strategy for memory summarization."""

    RELEVANCE = "relevance"
    RECENCY = "recency"
    IMPORTANCE = "importance"


@dataclass
@traces(spec="010", story="US-004", reqs=["REQ-010-005"])
class MemoryScope:
    """Scope filter for memory queries."""

    session_id: str | None = None
    task_id: str | None = None
    volk_id: str | None = None
    project_id: str | None = None
    portfolio_id: str | None = None
    date_from: datetime | None = None
    date_to: datetime | None = None

    @property
    def agent_id(self) -> str | None:
        """Deprecated alias for volk_id (identity migration → volk_id canonical)."""
        return self.volk_id


@traces(spec="010", story="US-001", reqs=["REQ-010-004"])
class MemoryEntry(BaseModel):
    """A memory entry stored by an Agent."""

    model_config = ConfigDict(populate_by_name=True)

    key: str = PydanticField(default_factory=lambda: str(uuid.uuid4()))
    content: str = ""
    summary: str | None = None
    tags: list[str] = PydanticField(default_factory=list)
    metadata: dict[str, Any] = PydanticField(default_factory=dict)
    classification: MemoryClassification = MemoryClassification.PRIVATE
    embedding: list[float] | None = None
    created_at: datetime = PydanticField(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = PydanticField(default_factory=lambda: datetime.now(UTC))
    volk_id: str = PydanticField(default="", alias="agent_id")
    session_id: str = ""
    task_id: str | None = None
    relevance_score: float | None = None
    token_count: int = 0

    @property
    def agent_id(self) -> str:
        """Deprecated alias for volk_id (identity migration → volk_id canonical)."""
        return self.volk_id

    def to_dict(self) -> dict[str, Any]:
        """Serialize to dict (backward compatibility)."""
        return self.model_dump()

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> MemoryEntry:
        """Deserialize from dict (backward compatibility).

        Handles stringified JSON for ``tags`` and ``metadata`` fields,
        and string ``token_count`` / ``classification`` values.
        """

        data = dict(data)
        for field_name in ("tags", "metadata"):
            value = data.get(field_name)
            if isinstance(value, str):
                data[field_name] = json.loads(value)
        if "token_count" in data and isinstance(data["token_count"], str):
            data["token_count"] = int(data["token_count"])
        if "classification" in data and isinstance(data["classification"], str):
            data["classification"] = MemoryClassification(data["classification"])
        return cls.model_validate(data)


@dataclass
@traces(spec="010", story="US-005", reqs=["REQ-010-023"])
class MemorySummary:
    """Result of memory summarization."""

    content: str = ""
    token_count: int = 0
    source_count: int = 0
    source_keys: list[str] = field(default_factory=list)
    strategy: SummarizationStrategy = SummarizationStrategy.RELEVANCE
    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))


# ============================================================================
# Token Counter Protocol (shared SPEC-010 / SPEC-011)
# ============================================================================


@traces(spec="010", story="US-003", reqs=["REQ-010-041"])
class TokenCounterProtocol(Protocol):
    """Provider-aware token counting abstraction."""

    @traces(spec="010", story="US-003", reqs=["REQ-010-041"])
    def count(self, text: str) -> int:
        """Count tokens in text."""
        ...

    @traces(spec="010", story="US-003", reqs=["REQ-010-042"])
    def truncate(self, text: str, max_tokens: int) -> str:
        """Truncate text to fit within max_tokens."""
        ...


# ============================================================================
# Memory Engine Protocol (SPEC-010)
# ============================================================================


@traces(spec="010", story="US-001", reqs=["REQ-010-001"])
class MemoryEngineProtocol(ComponentProtocol, Protocol):
    """Unified agent memory abstraction.

    Provides store, retrieve, search, summarize, and lifecycle
    operations over a dual-layer storage system (Redis hot / Neo4j cold).
    """

    @traces(spec="010", story="US-001", reqs=["REQ-010-006", "REQ-010-007"])
    async def store(
        self,
        content: str,
        *,
        volk_id: str,
        session_id: str,
        key: str | None = None,
        tags: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        classification: MemoryClassification = MemoryClassification.PRIVATE,
        task_id: str | None = None,
    ) -> MemoryEntry:
        """Store a memory entry. Returns the created entry with generated key."""
        ...

    @traces(spec="010", story="US-001", reqs=["REQ-010-011", "REQ-010-014"])
    async def retrieve(
        self,
        key: str,
        *,
        include_metadata: bool = True,
    ) -> MemoryEntry | None:
        """Retrieve a memory entry by exact key."""
        ...

    @traces(
        spec="010",
        story="US-004",
        reqs=["REQ-010-015", "REQ-010-017", "REQ-010-019", "REQ-010-039"],
    )
    async def search(
        self,
        query: str,
        *,
        scope: MemoryScope | None = None,
        limit: int = 10,
        min_similarity: float = 0.7,
        max_entries: int | None = None,
        max_age_seconds: int | None = None,
    ) -> list[MemoryEntry]:
        """Semantic search over memory entries.

        Args:
            query: Free-text query to match against memory content.
            scope: Optional scope filter (session/volk/task/date range).
            limit: Maximum results to return after similarity scoring.
            min_similarity: Minimum cosine similarity for a candidate.
            max_entries: Pre-filter — maximum candidate entries to retrieve
                from storage before embedding scoring. Keeps recall latency
                bounded when the hot store is large.
            max_age_seconds: Pre-filter — only retrieve entries created within
                this many seconds of now. Applied before similarity scoring.
        """
        ...

    @traces(spec="010", story="US-005", reqs=["REQ-010-020", "REQ-010-022"])
    async def summarize(
        self,
        context_id: str,
        *,
        max_tokens: int = 2048,
        strategy: SummarizationStrategy = SummarizationStrategy.RELEVANCE,
    ) -> MemorySummary:
        """Compress memories for context_id into a token-bounded summary."""
        ...

    @traces(spec="010", story="US-006", reqs=["REQ-010-050"])
    async def update(
        self,
        key: str,
        *,
        content: str | None = None,
        tags: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        classification: MemoryClassification | None = None,
    ) -> MemoryEntry:
        """Update an existing memory entry. Only provided fields are modified."""
        ...

    async def delete(self, key: str) -> bool:
        """Delete a memory entry by key. Returns True if found and deleted."""
        ...

    @traces(spec="010", story="US-002", reqs=["REQ-010-032"])
    async def migrate_to_cold(self, session_id: str) -> int:
        """Migrate all hot memories for session to cold storage. Returns count."""
        ...


# Compat alias matching the core.interfaces naming convention (P2.2).
IMemoryEngine = MemoryEngineProtocol

__all__ = [
    "MemoryClassification",
    "SummarizationStrategy",
    "MemoryScope",
    "MemoryEntry",
    "MemorySummary",
    "TokenCounterProtocol",
    "MemoryEngineProtocol",
    "IMemoryEngine",
]