# app/models/memory/working_memory.py """ Working Memory database model. Stores ephemeral key-value data for active sessions with TTL support. Used as database backup when Redis is unavailable. """ from sqlalchemy import Column, DateTime, Enum, Index, String from sqlalchemy.dialects.postgresql import JSONB from app.models.base import Base, TimestampMixin, UUIDMixin from .enums import ScopeType class WorkingMemory(Base, UUIDMixin, TimestampMixin): """ Working memory storage table. Provides database-backed working memory as fallback when Redis is unavailable. Supports TTL-based expiration. """ __tablename__ = "working_memory" # Scoping scope_type: Column[ScopeType] = Column( Enum(ScopeType), nullable=False, index=True, ) scope_id = Column(String(255), nullable=False, index=True) # Key-value storage key = Column(String(255), nullable=False) value = Column(JSONB, nullable=False) # TTL support expires_at = Column(DateTime(timezone=True), nullable=True, index=True) __table_args__ = ( # Primary lookup: scope + key Index( "ix_working_memory_scope_key", "scope_type", "scope_id", "key", unique=True, ), # For cleanup of expired entries Index("ix_working_memory_expires", "expires_at"), # For listing all keys in a scope Index("ix_working_memory_scope_list", "scope_type", "scope_id"), ) def __repr__(self) -> str: return f""