61 lines
2.4 KiB
Python
61 lines
2.4 KiB
Python
"""Tenant-scoped memory namespace and access policy for AIOS V2R1."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import PurePosixPath
|
|
from typing import Literal
|
|
|
|
from .tenant_context import ScopeError, ValidationError, require_ident
|
|
|
|
|
|
MemoryKind = Literal["session", "profile", "cell", "tenant", "shared_artifact"]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MemoryScope:
|
|
tenant_id: str
|
|
cell_id: str
|
|
profile_id: str
|
|
session_id: str | None
|
|
kind: MemoryKind
|
|
|
|
def __post_init__(self) -> None:
|
|
require_ident("tenant_id", self.tenant_id, tenant=True)
|
|
for name in ("cell_id", "profile_id"):
|
|
require_ident(name, getattr(self, name))
|
|
if self.session_id is not None:
|
|
require_ident("session_id", self.session_id)
|
|
if self.kind not in {"session", "profile", "cell", "tenant", "shared_artifact"}:
|
|
raise ValidationError("invalid memory kind")
|
|
if self.kind == "session" and self.session_id is None:
|
|
raise ValidationError("session memory requires session_id")
|
|
|
|
@property
|
|
def namespace(self) -> str:
|
|
parts = ["tenant", self.tenant_id, "cell", self.cell_id]
|
|
if self.kind in {"profile", "session"}:
|
|
parts.extend(["profile", self.profile_id])
|
|
if self.kind == "session":
|
|
parts.extend(["session", str(self.session_id)])
|
|
parts.extend(["memory", self.kind])
|
|
return str(PurePosixPath(*parts))
|
|
|
|
def assert_access(self, requested: "MemoryScope", *, write: bool = False) -> None:
|
|
if self.tenant_id != requested.tenant_id:
|
|
raise ScopeError("cross-tenant memory access denied")
|
|
if self.cell_id != requested.cell_id:
|
|
raise ScopeError("cross-cell memory access denied")
|
|
if requested.kind == "shared_artifact":
|
|
if write:
|
|
raise ScopeError("shared artifacts are immutable")
|
|
return
|
|
if self.profile_id != requested.profile_id and requested.kind in {"profile", "session"}:
|
|
raise ScopeError("cross-profile memory access denied")
|
|
if requested.kind == "session" and self.session_id != requested.session_id:
|
|
raise ScopeError("cross-session memory access denied")
|
|
|
|
|
|
def qualify_memory_key(scope: MemoryScope, key: str) -> str:
|
|
key = require_ident("memory_key", key)
|
|
return f"{scope.namespace}/{key}"
|