diff --git a/backend/apps/config_app.py b/backend/apps/config_app.py index f2d23b4079..6db0ee7964 100644 --- a/backend/apps/config_app.py +++ b/backend/apps/config_app.py @@ -41,6 +41,8 @@ from apps.aidp_app import router as aidp_router from apps.cas_app import router as cas_router from apps.quota_app import tenant_quota_router, platform_quota_router +from apps.memory_record_app import router as memory_record_router +from apps.memory_dreaming_app import router as memory_dreaming_router from consts.const import IS_SPEED_MODE from services.prompt_template_service import sync_system_default_prompt_template @@ -107,3 +109,5 @@ async def sync_default_prompt_template_on_startup(): app.include_router(aidp_router) app.include_router(tenant_quota_router) app.include_router(platform_quota_router) +app.include_router(memory_record_router) +app.include_router(memory_dreaming_router) diff --git a/backend/apps/memory_dreaming_app.py b/backend/apps/memory_dreaming_app.py new file mode 100644 index 0000000000..d20d660244 --- /dev/null +++ b/backend/apps/memory_dreaming_app.py @@ -0,0 +1,54 @@ +"""Manual Dreaming run and audit endpoints.""" + +from http import HTTPStatus +from typing import Annotated, Optional + +from fastapi import APIRouter, Header, HTTPException, Query +from pydantic import BaseModel, Field + +from services.memory_dreaming_service import ( + DreamingRunError, + get_memory_dreaming_service, +) +from utils.auth_utils import get_current_user_id + +router = APIRouter(prefix="/memory/dreaming", tags=["memory-dreaming"]) + + +class DreamingRunRequest(BaseModel): + agent_id: str = Field(..., min_length=1) + + +@router.post("/run") +def run_dreaming( + payload: DreamingRunRequest, + authorization: Annotated[Optional[str], Header()] = None, +): + user_id, tenant_id = get_current_user_id(authorization) + try: + return get_memory_dreaming_service().run( + tenant_id=tenant_id, + user_id=user_id, + agent_id=payload.agent_id, + ) + except DreamingRunError as exc: + raise HTTPException( + status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(exc) + ) from exc + + +@router.get("/audit") +def list_dreaming_audits( + authorization: Annotated[Optional[str], Header()] = None, + agent_id: Annotated[Optional[str], Query()] = None, + run_id: Annotated[Optional[int], Query(ge=1)] = None, + limit: Annotated[int, Query(ge=1, le=500)] = 100, +): + user_id, tenant_id = get_current_user_id(authorization) + return get_memory_dreaming_service().list_audits( + tenant_id, + user_id, + agent_id=agent_id, + run_id=run_id, + limit=limit, + ) diff --git a/backend/database/db_models.py b/backend/database/db_models.py index 6291b7aa8b..053864fabd 100644 --- a/backend/database/db_models.py +++ b/backend/database/db_models.py @@ -760,6 +760,111 @@ class MemoryUserConfig(TableBase): config_value = Column(String(10000), doc="the value of the config") +class MemoryRecord(TableBase): + """Authoritative tenant/user/agent memory row.""" + + __tablename__ = "memory_records_t" + __table_args__ = ( + Index("idx_memory_records_tenant", "tenant_id"), + Index("idx_memory_records_user", "tenant_id", "user_id"), + Index("idx_memory_records_agent", "tenant_id", "user_id", "agent_id", "conversation_id"), + Index("idx_memory_records_idempotency", "tenant_id", "idempotency_key"), + Index("idx_memory_records_status", "tenant_id", "user_id", "layer", "status"), + {"schema": SCHEMA}, + ) + + memory_id = Column(Integer, primary_key=True, nullable=False, autoincrement=True) + tenant_id = Column(String(100), nullable=False) + user_id = Column(String(100), nullable=False) + agent_id = Column(String(100)) + conversation_id = Column(String(100)) + layer = Column(String(30), nullable=False) + memory_type = Column(String(30)) + status = Column(String(30), nullable=False, default="active") + content = Column(Text, nullable=False) + concept_tags = Column(ARRAY(Text)) + es_index_name = Column(String(255)) + idempotency_key = Column(String(128), nullable=False) + recall_count = Column(Integer, nullable=False, default=0) + daily_count = Column(Integer, nullable=False, default=0) + grounded_count = Column(Integer, nullable=False, default=0) + last_recalled_at = Column(TIMESTAMP(timezone=False)) + query_hashes = Column(ARRAY(Text)) + recall_days = Column(ARRAY(Text)) + light_hits = Column(Integer, nullable=False, default=0) + rem_hits = Column(Integer, nullable=False, default=0) + last_light_at = Column(TIMESTAMP(timezone=False)) + last_rem_at = Column(TIMESTAMP(timezone=False)) + + +class MemoryRetrievalHit(TableBase): + """Append-only recall evidence consumed by Dreaming.""" + + __tablename__ = "memory_retrieval_hits_t" + __table_args__ = ( + Index("idx_memory_retrieval_hits_memory", "memory_id", "occurred_at"), + Index( + "idx_memory_retrieval_hits_tenant_user_agent", + "tenant_id", + "user_id", + "agent_id", + "day", + ), + {"schema": SCHEMA}, + ) + + hit_id = Column(Integer, primary_key=True, nullable=False, autoincrement=True) + tenant_id = Column(String(100)) + user_id = Column(String(100)) + agent_id = Column(String(100)) + conversation_id = Column(String(100)) + memory_id = Column(Integer) + query_text = Column(Text) + query_hash = Column(String(128)) + retrieval_score = Column(Numeric(38, 18)) + source = Column(String(100), nullable=False, default="nexent") + occurred_at = Column(TIMESTAMP(timezone=False), nullable=False, server_default=func.now()) + day = Column(String(100)) + grounded = Column(Boolean, nullable=False, default=False) + + +class MemoryDreamingAudit(TableBase): + """One durable audit row per manual Dreaming run.""" + + __tablename__ = "memory_dreaming_audit_t" + __table_args__ = ( + Index( + "idx_memory_dreaming_audit_scope", + "tenant_id", + "user_id", + "agent_id", + "started_at", + ), + {"schema": SCHEMA}, + ) + + run_id = Column( + BigInteger, + Sequence("memory_dreaming_audit_t_run_id_seq", schema=SCHEMA), + primary_key=True, + nullable=False, + ) + tenant_id = Column(String(100), nullable=False) + user_id = Column(String(100), nullable=False) + agent_id = Column(String(100), nullable=False) + trigger_source = Column(String(30), nullable=False, default="manual") + status = Column(String(30), nullable=False, default="running") + current_phase = Column(String(30)) + started_at = Column(TIMESTAMP(timezone=False), nullable=False, server_default=func.now()) + finished_at = Column(TIMESTAMP(timezone=False)) + light_count = Column(Integer, nullable=False, default=0) + rem_count = Column(Integer, nullable=False, default=0) + promoted_count = Column(Integer, nullable=False, default=0) + deferred_count = Column(Integer, nullable=False, default=0) + result_json = Column(JSONB) + error = Column(Text) + + class McpRecord(TableBase): """ MCP (Model Context Protocol) records table @@ -1093,32 +1198,6 @@ class SkillRepository(TableBase): doc="Listing status: not_shared / pending_review / rejected / shared") -class SkillRepository(TableBase): - """ - Skill repository (marketplace) table. Frozen snapshot of a shared skill for installation. - """ - __tablename__ = "ag_skill_repository_t" - __table_args__ = {"schema": SCHEMA} - - skill_repository_id = Column(BigInteger, Sequence("ag_skill_repository_t_skill_repository_id_seq", schema=SCHEMA), - primary_key=True, nullable=False, doc="Skill repository listing ID, unique primary key") - publisher_tenant_id = Column(String(100), nullable=False, doc=_PUBLISHER_TENANT_ID_DOC) - publisher_user_id = Column(String(100), nullable=False, doc=_PUBLISHER_USER_ID_DOC) - skill_id = Column(Integer, nullable=False, doc="Source skill ID from ag_skill_info_t") - name = Column(String(100), nullable=False, doc="Skill name for display and search") - description = Column(Text, doc="Skill description") - source = Column(String(30), doc="Skill source") - submitted_by = Column(String(100), doc="Submitter email when listing enters pending_review") - category_id = Column(Integer, doc="Optional marketplace category ID") - tags = Column(ARRAY(Text), doc="Marketplace tags") - icon = Column(String(100), doc="Marketplace card icon (emoji or URL)") - downloads = Column(Integer, default=0, doc="Marketplace install count for card display") - skill_info_json = Column(JSONB, nullable=False, doc="Frozen skill metadata snapshot") - skill_zip_base64 = Column(Text, nullable=False, doc="Frozen skill ZIP payload encoded as base64") - status = Column(String(30), default="not_shared", - doc="Listing status: not_shared / pending_review / rejected / shared") - - class UserTokenInfo(TableBase): """ User token (AK/SK) information table diff --git a/backend/database/memory_dreaming_db.py b/backend/database/memory_dreaming_db.py new file mode 100644 index 0000000000..e982766d8b --- /dev/null +++ b/backend/database/memory_dreaming_db.py @@ -0,0 +1,132 @@ +"""Persistence and PostgreSQL advisory locking for manual Dreaming runs.""" + +from __future__ import annotations + +import hashlib +from contextlib import contextmanager +from datetime import datetime, timezone +from typing import Any, Dict, Iterator, List, Optional + +from sqlalchemy import text + +from .client import get_db_session +from .db_models import MemoryDreamingAudit + + +def advisory_lock_key(tenant_id: str, user_id: str, agent_id: str) -> int: + digest = hashlib.sha256( + f"{tenant_id}:{user_id}:{agent_id}".encode("utf-8") + ).digest() + return int.from_bytes(digest[:8], "big", signed=True) + + +@contextmanager +def try_scope_lock(tenant_id: str, user_id: str, agent_id: str) -> Iterator[bool]: + """Hold a transaction-scoped advisory lock for the context lifetime.""" + with get_db_session() as session: + acquired = bool( + session.execute( + text("SELECT pg_try_advisory_xact_lock(:lock_key)"), + {"lock_key": advisory_lock_key(tenant_id, user_id, agent_id)}, + ).scalar() + ) + try: + yield acquired + session.commit() + except Exception: + session.rollback() + raise + + +def create_audit(tenant_id: str, user_id: str, agent_id: str) -> int: + with get_db_session() as session: + row = MemoryDreamingAudit( + tenant_id=tenant_id, + user_id=user_id, + agent_id=agent_id, + trigger_source="manual", + status="running", + current_phase="light", + ) + session.add(row) + session.commit() + return int(row.run_id) + + +def update_audit(run_id: int, values: Dict[str, Any]) -> bool: + allowed = { + "status", + "current_phase", + "finished_at", + "light_count", + "rem_count", + "promoted_count", + "deferred_count", + "result_json", + "error", + } + with get_db_session() as session: + row = ( + session.query(MemoryDreamingAudit) + .filter(MemoryDreamingAudit.run_id == run_id) + .first() + ) + if row is None: + return False + for key, value in values.items(): + if key in allowed: + setattr(row, key, value) + session.commit() + return True + + +def finish_audit(run_id: int, *, status: str, **values: Any) -> bool: + payload = { + **values, + "status": status, + "finished_at": datetime.now(timezone.utc).replace(tzinfo=None), + } + if status != "failed": + payload["current_phase"] = None + return update_audit(run_id, payload) + + +def list_audits( + tenant_id: str, + user_id: str, + *, + agent_id: Optional[str] = None, + run_id: Optional[int] = None, + limit: int = 100, +) -> List[Dict[str, Any]]: + with get_db_session() as session: + query = session.query(MemoryDreamingAudit).filter( + MemoryDreamingAudit.tenant_id == tenant_id, + MemoryDreamingAudit.user_id == user_id, + MemoryDreamingAudit.delete_flag == "N", + ) + if agent_id is not None: + query = query.filter(MemoryDreamingAudit.agent_id == agent_id) + if run_id is not None: + query = query.filter(MemoryDreamingAudit.run_id == run_id) + rows = query.order_by(MemoryDreamingAudit.run_id.desc()).limit(limit).all() + return [ + { + "run_id": row.run_id, + "tenant_id": row.tenant_id, + "user_id": row.user_id, + "agent_id": row.agent_id, + "trigger_source": row.trigger_source, + "status": row.status, + "current_phase": row.current_phase, + "started_at": row.started_at.isoformat() if row.started_at else None, + "finished_at": row.finished_at.isoformat() if row.finished_at else None, + "light_count": row.light_count, + "rem_count": row.rem_count, + "promoted_count": row.promoted_count, + "deferred_count": row.deferred_count, + "result": row.result_json, + "error": row.error, + } + for row in rows + ] diff --git a/backend/database/memory_retrieval_hit_db.py b/backend/database/memory_retrieval_hit_db.py index 9028ad41bf..d69b512905 100644 --- a/backend/database/memory_retrieval_hit_db.py +++ b/backend/database/memory_retrieval_hit_db.py @@ -186,6 +186,47 @@ def aggregate_memory_stats( return out +def aggregate_dreaming_stats( + tenant_id: str, + user_id: str, + agent_id: str, + *, + since: datetime, +) -> List[Dict[str, Any]]: + """Return complete Light/Deep evidence for one isolation scope.""" + hits = list_hits_for_user(tenant_id, user_id, since=since, limit=10000) + grouped: Dict[int, Dict[str, Any]] = {} + for hit in hits: + if str(hit.get("agent_id")) != str(agent_id) or hit.get("memory_id") is None: + continue + memory_id = int(hit["memory_id"]) + entry = grouped.setdefault( + memory_id, + { + "memory_id": memory_id, + "hit_count": 0, + "grounded_count": 0, + "days": set(), + "query_hashes": set(), + "total_retrieval_score": 0.0, + "last_recalled_at": None, + }, + ) + entry["hit_count"] += 1 + entry["grounded_count"] += int(bool(hit.get("grounded"))) + if hit.get("day"): + entry["days"].add(str(hit["day"])) + if hit.get("query_hash"): + entry["query_hashes"].add(str(hit["query_hash"])) + entry["total_retrieval_score"] += float(hit.get("retrieval_score") or 0) + occurred_at = hit.get("occurred_at") + if occurred_at and ( + entry["last_recalled_at"] is None or occurred_at > entry["last_recalled_at"] + ): + entry["last_recalled_at"] = occurred_at + return list(grouped.values()) + + def delete_hits_before(cutoff: datetime) -> int: """Delete hit rows older than ``cutoff`` (housekeeping).""" with get_db_session() as session: @@ -223,4 +264,4 @@ def _hit_to_dict(row: MemoryRetrievalHit) -> Dict[str, Any]: "occurred_at": row.occurred_at, "day": row.day, "grounded": bool(row.grounded), - } \ No newline at end of file + } diff --git a/backend/services/memory_context_service.py b/backend/services/memory_context_service.py index b60e207859..2f4aed3307 100644 --- a/backend/services/memory_context_service.py +++ b/backend/services/memory_context_service.py @@ -23,8 +23,8 @@ MemorySearchRequest, MemorySearchResult, PipelineConfig, - RetrievalPipeline, ) +from nexent.memory.retrieval import RetrievalPipeline from nexent.memory.policy import MemoryRetrievalPolicy from consts.const import ( diff --git a/backend/services/memory_dreaming_scheduler.py b/backend/services/memory_dreaming_scheduler.py index 4f56345236..d6b0e4855e 100644 --- a/backend/services/memory_dreaming_scheduler.py +++ b/backend/services/memory_dreaming_scheduler.py @@ -1,411 +1,12 @@ -"""Dreaming consolidation runner for the Memory Architecture (Phase 2). +"""Compatibility facade for the former Phase 2 Dreaming placeholder.""" -Dreaming promotes agent short-term memories into user long-term memory. It -runs in three phases: +try: + from services.memory_dreaming_service import get_memory_dreaming_service +except ImportError: # package-style unit-test imports + from .memory_dreaming_service import get_memory_dreaming_service -1. **Light Sleep** - aggregate ``memory_retrieval_hits_t`` rows into the - per-memory ``light_hits`` counter and ``recall_count`` / - ``recall_days`` / ``query_hashes`` columns. -2. **REM Sleep** - extract repeating concepts / patterns, write concept - tags to ``memory_records_t``. The phase is implemented as a lightweight - keyword-frequency pass; LLM-driven concept extraction can be wired in - later without changing the public API. -3. **Deep Sleep** - select eligible agent memories and promote them to - ``user`` long-term memory using the documented scoring formula - (frequency / relevance / diversity / recency / consolidation / - concept + phase boost). -Promotion thresholds and weights live in ``consts.const``. The phases are -exposed as standalone ``run_light_sleep`` / ``run_rem_sleep`` / -``run_deep_sleep`` functions plus the aggregate ``run_once`` so callers -can trigger a single pass per tenant on demand (e.g. from a future agent -timer). The module deliberately does **not** ship an internal scheduler, -background thread, or cron expression: agent-driven scheduling will be -added in a later phase, and we want to avoid having to coordinate cron, -lock watchdog and lifecycle here before the agent timer feature lands. -""" - -from __future__ import annotations - -import logging -import math -import time -from collections import Counter -from datetime import datetime, timedelta -from typing import Any, Dict, List, Optional, Sequence, Set - -from consts.const import ( - AGENT_SHORT_TERM_HALF_LIFE_DAYS, - LIGHT_SLEEP_WINDOW_DAYS, - MIN_PROMOTION_SCORE, - MIN_RECALL_COUNT, - MIN_UNIQUE_QUERIES, - RECENCY_HALF_LIFE_DAYS, -) -from database import memory_record_db, memory_retrieval_hit_db -from services.memory_record_service import ( - MemoryRecordError, - get_memory_record_service, -) - - -logger = logging.getLogger("memory_dreaming_scheduler") - - -# --------------------------------------------------------------------------- -# Scoring helpers -# --------------------------------------------------------------------------- - - -def _clamp01(value: float) -> float: - if value < 0.0: - return 0.0 - if value > 1.0: - return 1.0 - return value - - -def _frequency(recall_count: int, daily_count: int, grounded_count: int) -> float: - """Log-scaled accumulation of recall signals.""" - signal = max(0, recall_count) + max(0, daily_count) + max(0, grounded_count) - return _clamp01(math.log1p(signal) / math.log1p(10)) - - -def _relevance(hit_count: int, total_score: float) -> float: - """Average retrieval score across hits, clamped to [0, 1].""" - if hit_count <= 0: - return 0.0 - return _clamp01(total_score / max(1, hit_count)) - - -def _diversity(unique_queries: int) -> float: - """Smoothed saturation in [0, 1].""" - return _clamp01(math.log1p(unique_queries) / math.log1p(5)) - - -def _recency(last_recalled_at: Optional[datetime]) -> float: - """Exponential decay based on ``RECENCY_HALF_LIFE_DAYS``.""" - if last_recalled_at is None: - return 0.0 - delta_days = (datetime.utcnow() - last_recalled_at).total_seconds() / 86400.0 - if delta_days < 0: - delta_days = 0 - half_life = max(1, RECENCY_HALF_LIFE_DAYS) - return _clamp01(math.pow(0.5, delta_days / half_life)) - - -def _consolidation(light_hits: int, rem_hits: int) -> float: - """Boost when both Light and REM phases have seen the memory.""" - combined = max(0, light_hits) + max(0, rem_hits) - return _clamp01(math.log1p(combined) / math.log1p(6)) - - -def _concept(concept_tags: Sequence[str]) -> float: - """Higher when concept tags have been attached.""" - if not concept_tags: - return 0.0 - return _clamp01(math.log1p(len(concept_tags)) / math.log1p(8)) - - -# Weights reflect ``openclaw_dreaming.md`` §深眠阶段评分体系. -_PROMOTION_WEIGHTS: Dict[str, float] = { - "relevance": 0.30, - "frequency": 0.24, - "diversity": 0.15, - "recency": 0.15, - "consolidation": 0.10, - "concept": 0.06, -} - - -def _normalize_weights(weights: Dict[str, float]) -> Dict[str, float]: - total = sum(weights.values()) or 1.0 - return {key: value / total for key, value in weights.items()} - - -def _phase_boost(light_hits: int, rem_hits: int) -> float: - """PhaseBoost from the design doc; kept small to avoid runaway scores.""" - if light_hits <= 0 or rem_hits <= 0: - return 0.0 - return _clamp01(min(0.05, light_hits * 0.01 + rem_hits * 0.01)) - - -def compute_promotion_score(record: Dict[str, Any]) -> float: - """Compute the composite score for a single memory record.""" - recall_count = int(record.get("recall_count") or 0) - daily_count = int(record.get("daily_count") or 0) - grounded_count = int(record.get("grounded_count") or 0) - light_hits = int(record.get("light_hits") or 0) - rem_hits = int(record.get("rem_hits") or 0) - last_recalled_at = record.get("last_recalled_at") - query_hashes = record.get("query_hashes") or [] - - metrics = { - "frequency": _frequency(recall_count, daily_count, grounded_count), - "relevance": _relevance(recall_count, 1.0), - "diversity": _diversity(len(query_hashes)), - "recency": _recency(last_recalled_at), - "consolidation": _consolidation(light_hits, rem_hits), - "concept": _concept(record.get("concept_tags") or []), - } - weights = _normalize_weights(_PROMOTION_WEIGHTS) - score = sum(metrics[key] * weights[key] for key in metrics) - score += _phase_boost(light_hits, rem_hits) - return _clamp01(score) - - -# --------------------------------------------------------------------------- -# Phase runners -# --------------------------------------------------------------------------- - - -def run_light_sleep( - *, - tenant_id: str, - user_id: str, - agent_id: Optional[str] = None, - window_days: int = LIGHT_SLEEP_WINDOW_DAYS, -) -> int: - """Aggregate recent hits into memory row counters. - - Returns the number of memory rows touched. - """ - since = datetime.utcnow() - timedelta(days=max(1, window_days)) - stats = memory_retrieval_hit_db.aggregate_memory_stats( - tenant_id, - user_id=user_id, - agent_id=agent_id, - since=since, - ) - touched = 0 - for entry in stats: - memory_id = entry["memory_id"] - # Last hit day is the most recent value in the per-memory hit set. - last_day = max(entry["days"]) if entry["days"] else None - last_recalled_at = ( - datetime.fromisoformat(last_day) if last_day else None - ) - memory_record_db.update_memory_record( - memory_id, - tenant_id, - { - "recall_count": entry["hit_count"], - "grounded_count": entry["grounded_count"], - "query_hashes": sorted(entry["query_hashes"]), - "recall_days": sorted(entry["days"]), - "last_recalled_at": last_recalled_at, - }, - ) - memory_record_db.apply_dreaming_phase( - memory_id, tenant_id, phase="light" - ) - touched += 1 - return touched - - -_KEYWORD_STOPWORDS: Set[str] = { - "the", "a", "an", "and", "or", "but", "is", "are", "was", "were", - "be", "been", "being", "have", "has", "had", "do", "does", "did", - "of", "in", "on", "at", "by", "for", "with", "to", "from", - "i", "you", "he", "she", "it", "we", "they", - "的", "了", "是", "在", "和", "与", "及", "或", "我", "你", "他", "她", "它", - "我们", "你们", "他们", "这", "那", "这个", "那个", -} - - -def _tokenize(text: str) -> List[str]: - return [ - token.strip().lower() - for token in text.replace("\n", " ").split() - if token.strip() and token.strip().lower() not in _KEYWORD_STOPWORDS - ] - - -def run_rem_sleep( - *, - tenant_id: str, - user_id: str, - agent_id: Optional[str] = None, - max_keywords: int = 5, -) -> int: - """Extract concept tags from frequently appearing tokens. - - Returns the number of memory rows whose ``concept_tags`` were updated. - """ - rows = memory_record_db.list_memory_records( - tenant_id, - user_id=user_id, - agent_id=agent_id, - layer="agent", - memory_type="short_term", - status="active", - limit=500, - ) - touched = 0 - for row in rows: - tokens = _tokenize(row.get("content", "")) - if not tokens: - continue - counter = Counter(tokens) - top = [token for token, _ in counter.most_common(max_keywords)] - if not top: - continue - existing = list(row.get("concept_tags") or []) - merged = list(dict.fromkeys(existing + top))[:max_keywords] - memory_record_db.update_memory_record( - row["memory_id"], - tenant_id, - {"concept_tags": merged}, - ) - memory_record_db.apply_dreaming_phase( - row["memory_id"], tenant_id, phase="rem" - ) - touched += 1 - return touched - - -def run_deep_sleep( - *, - tenant_id: str, - user_id: str, - agent_id: Optional[str] = None, - min_score: float = MIN_PROMOTION_SCORE, - min_recall_count: int = MIN_RECALL_COUNT, - min_unique_queries: int = MIN_UNIQUE_QUERIES, -) -> List[Dict[str, Any]]: - """Promote agent memories that pass the promotion thresholds. - - Returns the list of promotion results (``memory_id``, ``score``, ``event``). - """ - eligible = memory_record_db.list_memories_for_dreaming( - tenant_id, - user_id=user_id, - layer="agent", - min_recall_count=min_recall_count, - window_days=LIGHT_SLEEP_WINDOW_DAYS, +def run_once(*, tenant_id: str, user_id: str, agent_id: str, **kwargs): + return get_memory_dreaming_service().run( + tenant_id=tenant_id, user_id=user_id, agent_id=agent_id, **kwargs ) - promoted: List[Dict[str, Any]] = [] - service = get_memory_record_service() - for row in eligible: - query_hashes = row.get("query_hashes") or [] - if len(set(query_hashes)) < min_unique_queries: - continue - score = compute_promotion_score(row) - if score < min_score: - continue - try: - service.create_memory( - tenant_id=tenant_id, - user_id=user_id, - content=row.get("content", ""), - layer="user", - memory_type="long_term", - agent_id=row.get("agent_id"), - conversation_id=row.get("conversation_id"), - concept_tags=row.get("concept_tags") or [], - idempotency_key=f"dreaming:{row['memory_id']}", - created_by="dreaming", - actor="dreaming", - ) - except MemoryRecordError as exc: - logger.warning( - "dreaming promotion skipped for %s: %s", row["memory_id"], exc - ) - continue - memory_record_db.apply_dreaming_phase( - row["memory_id"], tenant_id, phase="rem" - ) - promoted.append( - { - "memory_id": row["memory_id"], - "score": score, - "event": "PROMOTE", - } - ) - return promoted - - -# --------------------------------------------------------------------------- -# Manual entry points -# --------------------------------------------------------------------------- - - -def run_once(*, timeout_seconds: int = 1800) -> Dict[str, Any]: - """Execute one full Dreaming cycle across known tenants. - - This function is the single manual entry point: callers (e.g. an agent - timer introduced later) invoke ``run_once`` whenever they want a fresh - pass. Phase 2 does not ship a scheduler; the function is intentionally - synchronous and idempotent so it can be re-invoked safely. - - Args: - timeout_seconds: Soft cap on wall-clock runtime per call. Iteration - stops once the deadline is reached; partial state is returned. - - Returns: - Summary dict with tenant count, light/rem rows touched, and the - list of promotion events. - """ - started = time.time() - deadline = started + max(60, timeout_seconds) - - # ``list_distinct_tenants`` is intentionally conservative: we only run - # dreaming over tenants that have actually touched memory recently. - tenants = list_distinct_tenants() - summary: Dict[str, Any] = { - "tenants": len(tenants), - "light_rows": 0, - "rem_rows": 0, - "promotions": [], - } - - for tenant_id, user_id in tenants: - if time.time() >= deadline: - logger.warning("Dreaming run hit timeout; aborting remaining tenants") - break - try: - light = run_light_sleep(tenant_id=tenant_id, user_id=user_id) - rem = run_rem_sleep(tenant_id=tenant_id, user_id=user_id) - deep = run_deep_sleep(tenant_id=tenant_id, user_id=user_id) - summary["light_rows"] += light - summary["rem_rows"] += rem - summary["promotions"].extend(deep) - except Exception: - logger.exception( - "Dreaming iteration failed for tenant=%s user=%s", - tenant_id, - user_id, - ) - - summary["elapsed_seconds"] = time.time() - started - return summary - - -def list_distinct_tenants() -> List[Any]: - """Return ``(tenant_id, user_id)`` tuples with recent memory activity. - - Implementation: distinct pairs from ``memory_retrieval_hits_t``. When - no hits exist (fresh deployments) this returns ``[]`` and Dreaming - becomes a no-op, which is the intended behavior. - """ - try: - from database.client import get_db_session - from database.db_models import MemoryRetrievalHit - from sqlalchemy import distinct - - with get_db_session() as session: - rows = ( - session.query( - distinct(MemoryRetrievalHit.tenant_id), - distinct(MemoryRetrievalHit.user_id), - ) - .filter( - MemoryRetrievalHit.tenant_id.isnot(None), - MemoryRetrievalHit.user_id.isnot(None), - ) - .all() - ) - return [(t, u) for t, u in rows if t and u] - except Exception: - logger.exception("list_distinct_tenants failed") - return [] \ No newline at end of file diff --git a/backend/services/memory_dreaming_service.py b/backend/services/memory_dreaming_service.py new file mode 100644 index 0000000000..cc13ff533c --- /dev/null +++ b/backend/services/memory_dreaming_service.py @@ -0,0 +1,247 @@ +"""Backend orchestration for the SDK Dreaming algorithm.""" + +from __future__ import annotations + +import logging +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional + +from consts.const import ( + LIGHT_SLEEP_WINDOW_DAYS, + MIN_PROMOTION_SCORE, + MIN_RECALL_COUNT, + MIN_UNIQUE_QUERIES, + RECENCY_HALF_LIFE_DAYS, +) +from database import memory_dreaming_db, memory_record_db, memory_retrieval_hit_db +from nexent.memory.dreaming import ( + DreamingThresholds, + build_candidate, + select_candidates, +) +from services.memory_record_service import get_memory_record_service + +logger = logging.getLogger("memory_dreaming_service") + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +class DreamingRunError(RuntimeError): + pass + + +class MemoryDreamingService: + def __init__(self, record_service: Any = None): + self.record_service = record_service or get_memory_record_service() + + def _run_light( + self, tenant_id: str, user_id: str, agent_id: str, window_days: int + ) -> Dict[int, Dict[str, Any]]: + stats = memory_retrieval_hit_db.aggregate_dreaming_stats( + tenant_id, + user_id, + agent_id, + since=_utcnow() - timedelta(days=max(1, window_days)), + ) + by_id = {int(item["memory_id"]): item for item in stats} + for item in stats: + memory_record_db.update_memory_record( + item["memory_id"], + tenant_id, + { + "recall_count": item["hit_count"], + "daily_count": len(item["days"]), + "grounded_count": item["grounded_count"], + "last_recalled_at": item["last_recalled_at"], + "query_hashes": sorted(item["query_hashes"]), + "recall_days": sorted(item["days"]), + }, + ) + memory_record_db.apply_dreaming_phase( + item["memory_id"], tenant_id, phase="light" + ) + return by_id + + def _run_rem( + self, + tenant_id: str, + user_id: str, + agent_id: str, + stats: Dict[int, Dict[str, Any]], + ) -> List[Any]: + records = memory_record_db.list_memory_records( + tenant_id, + user_id=user_id, + agent_id=agent_id, + layer="agent", + memory_type="short_term", + status="active", + limit=1000, + ) + candidates = [] + for record in records: + evidence = stats.get(int(record["memory_id"]), {}) + candidate = build_candidate( + record, float(evidence.get("total_retrieval_score") or 0) + ) + candidate.already_promoted = ( + memory_record_db.find_by_idempotency( + tenant_id, f"dreaming:{candidate.memory_id}" + ) + is not None + ) + memory_record_db.update_memory_record( + candidate.memory_id, + tenant_id, + {"concept_tags": candidate.concept_tags}, + ) + if not candidate.noise: + memory_record_db.apply_dreaming_phase( + candidate.memory_id, tenant_id, phase="rem" + ) + candidate.rem_hits += 1 + candidate.last_rem_at = _utcnow() + candidates.append(candidate) + return candidates + + def _promote(self, decisions: List[Any]) -> List[Dict[str, Any]]: + results = [] + for decision in decisions: + candidate = decision.candidate + if decision.promote: + created = self.record_service.create_memory( + tenant_id=candidate.tenant_id, + user_id=candidate.user_id, + agent_id=candidate.agent_id, + content=candidate.content, + layer="user", + memory_type="long_term", + concept_tags=candidate.concept_tags, + idempotency_key=f"dreaming:{candidate.memory_id}", + created_by="dreaming", + actor="dreaming", + ) + event = created.get("event", "ADD") + else: + event = "DEFER" + results.append( + { + "memory_id": candidate.memory_id, + "score": decision.score, + "event": event, + "reason": decision.reason, + "archive_suggested": decision.archive_suggested, + } + ) + return results + + def run( + self, + *, + tenant_id: str, + user_id: str, + agent_id: str, + window_days: int = LIGHT_SLEEP_WINDOW_DAYS, + min_score: float = MIN_PROMOTION_SCORE, + min_recall_count: int = MIN_RECALL_COUNT, + min_unique_queries: int = MIN_UNIQUE_QUERIES, + ) -> Dict[str, Any]: + if not tenant_id or not user_id or not agent_id: + raise DreamingRunError("tenant_id, user_id and agent_id are required") + run_id = memory_dreaming_db.create_audit(tenant_id, user_id, agent_id) + with memory_dreaming_db.try_scope_lock( + tenant_id, user_id, agent_id + ) as acquired: + if not acquired: + result = { + "run_id": run_id, + "status": "skipped", + "reason": "lock_busy", + } + memory_dreaming_db.finish_audit( + run_id, status="skipped", result_json=result + ) + return result + try: + stats = self._run_light(tenant_id, user_id, agent_id, window_days) + memory_dreaming_db.update_audit( + run_id, + {"current_phase": "rem", "light_count": len(stats)}, + ) + candidates = self._run_rem(tenant_id, user_id, agent_id, stats) + memory_dreaming_db.update_audit( + run_id, + {"current_phase": "deep", "rem_count": len(candidates)}, + ) + decisions = select_candidates( + candidates, + thresholds=DreamingThresholds( + min_score=min_score, + min_recall_count=min_recall_count, + min_unique_queries=min_unique_queries, + ), + recency_half_life_days=RECENCY_HALF_LIFE_DAYS, + ) + results = self._promote(decisions) + promoted_count = sum( + item["event"] in {"ADD", "UPDATE"} for item in results + ) + result = { + "run_id": run_id, + "status": "completed", + "light_count": len(stats), + "rem_count": len(candidates), + "promoted_count": promoted_count, + "deferred_count": len(results) - promoted_count, + "decisions": results, + } + memory_dreaming_db.finish_audit( + run_id, + status="completed", + light_count=len(stats), + rem_count=len(candidates), + promoted_count=promoted_count, + deferred_count=len(results) - promoted_count, + result_json=result, + ) + return result + except Exception as exc: + logger.exception( + "Dreaming failed for tenant=%s user=%s agent=%s run=%s", + tenant_id, + user_id, + agent_id, + run_id, + ) + error = f"{type(exc).__name__}: Dreaming phase failed" + memory_dreaming_db.finish_audit(run_id, status="failed", error=error) + raise DreamingRunError(error) from exc + + def list_audits( + self, + tenant_id: str, + user_id: str, + *, + agent_id: Optional[str] = None, + run_id: Optional[int] = None, + limit: int = 100, + ) -> List[Dict[str, Any]]: + return memory_dreaming_db.list_audits( + tenant_id, + user_id, + agent_id=agent_id, + run_id=run_id, + limit=limit, + ) + + +_service: Optional[MemoryDreamingService] = None + + +def get_memory_dreaming_service() -> MemoryDreamingService: + global _service + if _service is None: + _service = MemoryDreamingService() + return _service diff --git a/backend/utils/memory_utils.py b/backend/utils/memory_utils.py new file mode 100644 index 0000000000..3bac960770 --- /dev/null +++ b/backend/utils/memory_utils.py @@ -0,0 +1,13 @@ +"""Compatibility helper for agent cleanup paths on the new Memory system.""" + +from typing import Any, Dict + + +def build_memory_config(_tenant_id: str) -> Dict[str, Any]: + """Return an empty legacy config. + + The removed Mem0 functions still accept this argument at a few guarded + cleanup call sites. New Memory services resolve tenant model configuration + through backend services instead of this utility. + """ + return {} diff --git a/deploy/sql/init.sql b/deploy/sql/init.sql index 9d64ab2423..cc939d923c 100644 --- a/deploy/sql/init.sql +++ b/deploy/sql/init.sql @@ -718,3 +718,30 @@ FOR EACH ROW EXECUTE FUNCTION nexent.update_memory_retrieval_hits_update_time(); COMMENT ON TRIGGER update_memory_retrieval_hits_update_time_trigger ON nexent.memory_retrieval_hits_t IS 'Trigger to call update_memory_retrieval_hits_update_time function before each update on memory_retrieval_hits_t table'; +-- Manual Dreaming run audit. Scope concurrency uses PostgreSQL transaction +-- advisory locks, so no persistent lock row is required. +CREATE TABLE IF NOT EXISTS nexent.memory_dreaming_audit_t ( + run_id BIGSERIAL PRIMARY KEY, + tenant_id VARCHAR(100) NOT NULL, + user_id VARCHAR(100) NOT NULL, + agent_id VARCHAR(100) NOT NULL, + trigger_source VARCHAR(30) NOT NULL DEFAULT 'manual', + status VARCHAR(30) NOT NULL DEFAULT 'running', + current_phase VARCHAR(30), + started_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + finished_at TIMESTAMP, + light_count INTEGER NOT NULL DEFAULT 0, + rem_count INTEGER NOT NULL DEFAULT 0, + promoted_count INTEGER NOT NULL DEFAULT 0, + deferred_count INTEGER NOT NULL DEFAULT 0, + result_json JSONB, + error TEXT, + create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + created_by VARCHAR(100), + updated_by VARCHAR(100), + delete_flag VARCHAR(1) NOT NULL DEFAULT 'N' +); +CREATE INDEX IF NOT EXISTS idx_memory_dreaming_audit_scope + ON nexent.memory_dreaming_audit_t + (tenant_id, user_id, agent_id, started_at DESC); diff --git a/deploy/sql/migrations/v2.4.0_0723_add_memory_dreaming_audit.sql b/deploy/sql/migrations/v2.4.0_0723_add_memory_dreaming_audit.sql new file mode 100644 index 0000000000..4986a2b959 --- /dev/null +++ b/deploy/sql/migrations/v2.4.0_0723_add_memory_dreaming_audit.sql @@ -0,0 +1,33 @@ +-- Manual Dreaming run audit. Advisory locks are transaction-scoped and +-- therefore require no persistent lock table. +SET search_path TO nexent; +BEGIN; + +CREATE TABLE IF NOT EXISTS nexent.memory_dreaming_audit_t ( + run_id BIGSERIAL PRIMARY KEY, + tenant_id VARCHAR(100) NOT NULL, + user_id VARCHAR(100) NOT NULL, + agent_id VARCHAR(100) NOT NULL, + trigger_source VARCHAR(30) NOT NULL DEFAULT 'manual', + status VARCHAR(30) NOT NULL DEFAULT 'running', + current_phase VARCHAR(30), + started_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + finished_at TIMESTAMP, + light_count INTEGER NOT NULL DEFAULT 0, + rem_count INTEGER NOT NULL DEFAULT 0, + promoted_count INTEGER NOT NULL DEFAULT 0, + deferred_count INTEGER NOT NULL DEFAULT 0, + result_json JSONB, + error TEXT, + create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + created_by VARCHAR(100), + updated_by VARCHAR(100), + delete_flag VARCHAR(1) NOT NULL DEFAULT 'N' +); + +CREATE INDEX IF NOT EXISTS idx_memory_dreaming_audit_scope + ON nexent.memory_dreaming_audit_t + (tenant_id, user_id, agent_id, started_at DESC); + +COMMIT; diff --git a/sdk/nexent/memory/dreaming/__init__.py b/sdk/nexent/memory/dreaming/__init__.py new file mode 100644 index 0000000000..589f3f2bd8 --- /dev/null +++ b/sdk/nexent/memory/dreaming/__init__.py @@ -0,0 +1,23 @@ +"""Storage-independent Dreaming consolidation primitives.""" + +from .models import ( + DreamingCandidate, + DreamingDecision, + DreamingMetrics, + DreamingThresholds, +) +from .scoring import compute_metrics, score_candidate, select_candidates +from .service import analyze_rem_content, build_candidate + + +__all__ = [ + "DreamingCandidate", + "DreamingDecision", + "DreamingMetrics", + "DreamingThresholds", + "analyze_rem_content", + "build_candidate", + "compute_metrics", + "score_candidate", + "select_candidates", +] diff --git a/sdk/nexent/memory/dreaming/models.py b/sdk/nexent/memory/dreaming/models.py new file mode 100644 index 0000000000..1c46a3b506 --- /dev/null +++ b/sdk/nexent/memory/dreaming/models.py @@ -0,0 +1,58 @@ +"""Models shared by the three Dreaming phases.""" + +from __future__ import annotations + +from datetime import datetime +from typing import List, Optional + +from pydantic import BaseModel, Field + + +class DreamingCandidate(BaseModel): + memory_id: int + tenant_id: str + user_id: str + agent_id: str + content: str + recall_count: int = 0 + daily_count: int = 0 + grounded_count: int = 0 + total_retrieval_score: float = 0.0 + query_hashes: List[str] = Field(default_factory=list) + recall_days: List[str] = Field(default_factory=list) + concept_tags: List[str] = Field(default_factory=list) + light_hits: int = 0 + rem_hits: int = 0 + last_recalled_at: Optional[datetime] = None + last_light_at: Optional[datetime] = None + last_rem_at: Optional[datetime] = None + noise: bool = False + already_promoted: bool = False + + +class DreamingMetrics(BaseModel): + signal_count: int + context_diversity: int + frequency: float + relevance: float + query_diversity: float + recency: float + consolidation: float + conceptual_richness: float + phase_boost: float + + +class DreamingThresholds(BaseModel): + min_score: float = 0.72 + min_recall_count: int = 3 + min_unique_queries: int = 2 + include_promoted: bool = False + + +class DreamingDecision(BaseModel): + candidate: DreamingCandidate + metrics: DreamingMetrics + score: float + promote: bool + reason: str + archive_suggested: bool = False diff --git a/sdk/nexent/memory/dreaming/scoring.py b/sdk/nexent/memory/dreaming/scoring.py new file mode 100644 index 0000000000..9f69280c76 --- /dev/null +++ b/sdk/nexent/memory/dreaming/scoring.py @@ -0,0 +1,146 @@ +"""OpenClaw-compatible Deep Sleep scoring and deterministic selection.""" + +from __future__ import annotations + +import math +from datetime import datetime, timezone +from typing import Iterable, List, Optional + +from .models import ( + DreamingCandidate, + DreamingDecision, + DreamingMetrics, + DreamingThresholds, +) + + +WEIGHTS = { + "frequency": 0.24, + "relevance": 0.30, + "query_diversity": 0.15, + "recency": 0.15, + "consolidation": 0.10, + "conceptual_richness": 0.06, +} + + +def clamp_score(value: float) -> float: + return max(0.0, min(1.0, value)) + + +def _age_days(value: Optional[datetime], now: datetime) -> float: + if value is None: + return float("inf") + return max(0.0, (now - value).total_seconds() / 86400.0) + + +def _recency(value: Optional[datetime], now: datetime, half_life_days: float) -> float: + if value is None: + return 0.0 + decay_lambda = math.log(2) / max(1.0, half_life_days) + return clamp_score(math.exp(-decay_lambda * _age_days(value, now))) + + +def compute_metrics( + candidate: DreamingCandidate, + *, + now: Optional[datetime] = None, + recency_half_life_days: float = 14, +) -> DreamingMetrics: + now = now or datetime.now(timezone.utc).replace(tzinfo=None) + signal_count = max(0, candidate.recall_count) + max(0, candidate.daily_count) + max(0, candidate.grounded_count) + unique_queries = len(set(candidate.query_hashes)) + unique_days = len(set(candidate.recall_days)) + context_diversity = max(unique_queries, unique_days) + frequency = clamp_score(math.log1p(signal_count) / math.log1p(10)) + relevance = clamp_score(candidate.total_retrieval_score / max(1, signal_count)) + query_diversity = clamp_score(context_diversity / 5) + recency = _recency(candidate.last_recalled_at, now, recency_half_life_days) + + if unique_days == 0: + consolidation = 0.0 + elif unique_days == 1: + consolidation = 0.2 + else: + parsed_days = sorted(datetime.fromisoformat(day).date() for day in set(candidate.recall_days)) + span_days = (parsed_days[-1] - parsed_days[0]).days + spacing = clamp_score(math.log1p(unique_days - 1) / math.log1p(4)) + span = clamp_score(span_days / 7) + consolidation = max( + clamp_score(0.55 * spacing + 0.45 * span), + clamp_score(candidate.grounded_count / 3), + ) + + conceptual_richness = clamp_score(len(set(candidate.concept_tags)) / 6) + light_strength = clamp_score(math.log1p(max(0, candidate.light_hits)) / math.log1p(6)) + rem_strength = clamp_score(math.log1p(max(0, candidate.rem_hits)) / math.log1p(6)) + phase_boost = clamp_score( + 0.06 * light_strength * _recency(candidate.last_light_at, now, 14) + + 0.09 * rem_strength * _recency(candidate.last_rem_at, now, 14) + ) + return DreamingMetrics( + signal_count=signal_count, + context_diversity=context_diversity, + frequency=frequency, + relevance=relevance, + query_diversity=query_diversity, + recency=recency, + consolidation=consolidation, + conceptual_richness=conceptual_richness, + phase_boost=phase_boost, + ) + + +def score_candidate( + candidate: DreamingCandidate, + *, + now: Optional[datetime] = None, + recency_half_life_days: float = 14, +) -> tuple[float, DreamingMetrics]: + metrics = compute_metrics(candidate, now=now, recency_half_life_days=recency_half_life_days) + score = sum(getattr(metrics, name) * weight for name, weight in WEIGHTS.items()) + return clamp_score(score + metrics.phase_boost), metrics + + +def select_candidates( + candidates: Iterable[DreamingCandidate], + *, + thresholds: Optional[DreamingThresholds] = None, + now: Optional[datetime] = None, + recency_half_life_days: float = 14, +) -> List[DreamingDecision]: + thresholds = thresholds or DreamingThresholds() + decisions: List[DreamingDecision] = [] + for candidate in candidates: + score, metrics = score_candidate(candidate, now=now, recency_half_life_days=recency_half_life_days) + reason = "eligible" + promote = True + if candidate.noise: + promote, reason = False, "noise" + elif candidate.already_promoted and not thresholds.include_promoted: + promote, reason = False, "already_promoted" + elif score < thresholds.min_score: + promote, reason = False, "score_below_threshold" + elif metrics.signal_count < thresholds.min_recall_count: + promote, reason = False, "recall_below_threshold" + elif metrics.context_diversity < thresholds.min_unique_queries: + promote, reason = False, "diversity_below_threshold" + decisions.append( + DreamingDecision( + candidate=candidate, + metrics=metrics, + score=score, + promote=promote, + reason=reason, + archive_suggested=promote, + ) + ) + return sorted( + decisions, + key=lambda item: ( + not item.promote, + -item.score, + -item.metrics.signal_count, + item.candidate.memory_id, + ), + ) diff --git a/sdk/nexent/memory/dreaming/service.py b/sdk/nexent/memory/dreaming/service.py new file mode 100644 index 0000000000..2d67e623d5 --- /dev/null +++ b/sdk/nexent/memory/dreaming/service.py @@ -0,0 +1,63 @@ +"""Lightweight REM analysis and candidate construction.""" + +from __future__ import annotations + +import re +from typing import Any, Dict, List, Tuple + +from .models import DreamingCandidate + + +NOISE_PATTERNS = ( + r"\b(todo|today|temporary|this session|current task)\b", + r"(今日|今天|待办|临时|本轮|当前任务)", +) +CONCEPT_PATTERNS = { + "preference": (r"\b(prefer|preference|always use|likes?)\b", r"(偏好|喜欢|习惯|总是使用)"), + "persistent": (r"\b(always|persist|long[- ]term|stable)\b", r"(长期|稳定|持续|固定)"), + "build": (r"\b(build|compile|deploy|package)\b", r"(构建|编译|部署|打包)"), + "failure": (r"\b(error|failure|failed|exception|bug)\b", r"(错误|失败|异常|故障)"), + "transaction": (r"\b(transaction|commit|rollback|atomic)\b", r"(事务|提交|回滚|原子)"), + "routing": (r"\b(route|routing|gateway|proxy)\b", r"(路由|网关|代理)"), +} + + +def analyze_rem_content(content: str) -> Tuple[List[str], bool]: + normalized = " ".join(content.split()).lower() + noise = any(re.search(pattern, normalized, re.IGNORECASE) for pattern in NOISE_PATTERNS) + tags = [ + tag + for tag, patterns in CONCEPT_PATTERNS.items() + if any(re.search(pattern, normalized, re.IGNORECASE) for pattern in patterns) + ] + word_tags = re.findall(r"[\u4e00-\u9fff]{2,8}|[a-z][a-z0-9_-]{2,}", normalized) + for tag in word_tags: + if tag not in tags and len(tags) < 8: + tags.append(tag) + return tags, noise + + +def build_candidate(record: Dict[str, Any], total_retrieval_score: float) -> DreamingCandidate: + tags, noise = analyze_rem_content(str(record.get("content") or "")) + merged_tags = list(dict.fromkeys([*(record.get("concept_tags") or []), *tags])) + return DreamingCandidate( + memory_id=int(record["memory_id"]), + tenant_id=str(record["tenant_id"]), + user_id=str(record["user_id"]), + agent_id=str(record["agent_id"]), + content=str(record.get("content") or ""), + recall_count=int(record.get("recall_count") or 0), + daily_count=int(record.get("daily_count") or 0), + grounded_count=int(record.get("grounded_count") or 0), + total_retrieval_score=float(total_retrieval_score or 0), + query_hashes=list(record.get("query_hashes") or []), + recall_days=list(record.get("recall_days") or []), + concept_tags=merged_tags, + light_hits=int(record.get("light_hits") or 0), + rem_hits=int(record.get("rem_hits") or 0), + last_recalled_at=record.get("last_recalled_at"), + last_light_at=record.get("last_light_at"), + last_rem_at=record.get("last_rem_at"), + noise=noise, + already_promoted=bool(record.get("already_promoted", False)), + ) diff --git a/sdk/nexent/memory/memory_service.py b/sdk/nexent/memory/memory_service.py new file mode 100644 index 0000000000..f80dc27ad1 --- /dev/null +++ b/sdk/nexent/memory/memory_service.py @@ -0,0 +1,28 @@ +"""Legacy import bridge removed by the new Memory architecture. + +The target branch still imports these names from agent code. Keeping explicit +errors here lets applications start while preventing the removed Mem0/local-ES +implementation from silently appearing to persist data. Callers already +degrade on these exceptions; new code must use ``nexent.memory.service`` with +backend hooks. +""" + + +class LegacyMemoryApiRemoved(RuntimeError): + pass + + +async def add_memory_in_levels(**_kwargs): + raise LegacyMemoryApiRemoved( + "add_memory_in_levels was removed; use MemoryService.store_memory with a backend hook" + ) + + +async def search_memory_in_levels(**_kwargs): + raise LegacyMemoryApiRemoved( + "search_memory_in_levels was removed; use MemoryService.search_memory with a backend hook" + ) + + +async def clear_memory(**_kwargs): + raise LegacyMemoryApiRemoved("clear_memory was removed; use the backend memory record service") diff --git a/test/backend/apps/test_memory_dreaming_app.py b/test/backend/apps/test_memory_dreaming_app.py new file mode 100644 index 0000000000..1842c1aa04 --- /dev/null +++ b/test/backend/apps/test_memory_dreaming_app.py @@ -0,0 +1,77 @@ +from unittest.mock import MagicMock + +import pytest +from fastapi import HTTPException +from pydantic import ValidationError + +from apps import memory_dreaming_app + + +def test_ac009_missing_agent_id_is_rejected(): + with pytest.raises(ValidationError): + memory_dreaming_app.DreamingRunRequest() + + +def test_ac009_run_uses_authenticated_scope(monkeypatch): + service = MagicMock() + service.run.return_value = {"run_id": 1, "status": "completed"} + monkeypatch.setattr( + memory_dreaming_app, "get_memory_dreaming_service", lambda: service + ) + monkeypatch.setattr( + memory_dreaming_app, + "get_current_user_id", + lambda _authorization: ("user-1", "tenant-1"), + ) + result = memory_dreaming_app.run_dreaming( + memory_dreaming_app.DreamingRunRequest(agent_id="agent-1"), + authorization="Bearer token", + ) + assert result["status"] == "completed" + assert service.run.call_args.kwargs["tenant_id"] == "tenant-1" + assert service.run.call_args.kwargs["user_id"] == "user-1" + assert service.run.call_args.kwargs["agent_id"] == "agent-1" + + +def test_ac009_audit_uses_authenticated_scope(monkeypatch): + service = MagicMock() + service.list_audits.return_value = [{"run_id": 2}] + monkeypatch.setattr( + memory_dreaming_app, "get_memory_dreaming_service", lambda: service + ) + monkeypatch.setattr( + memory_dreaming_app, + "get_current_user_id", + lambda _authorization: ("user-2", "tenant-2"), + ) + result = memory_dreaming_app.list_dreaming_audits( + authorization="Bearer token", + agent_id="agent-2", + run_id=2, + limit=100, + ) + assert result == [{"run_id": 2}] + service.list_audits.assert_called_once_with( + "tenant-2", "user-2", agent_id="agent-2", run_id=2, limit=100 + ) + + +def test_ac008_service_failure_maps_to_500(monkeypatch): + service = MagicMock() + service.run.side_effect = memory_dreaming_app.DreamingRunError("failed") + monkeypatch.setattr( + memory_dreaming_app, "get_memory_dreaming_service", lambda: service + ) + monkeypatch.setattr( + memory_dreaming_app, + "get_current_user_id", + lambda _authorization: ("user", "tenant"), + ) + request = memory_dreaming_app.DreamingRunRequest(agent_id="agent") + with pytest.raises(HTTPException) as exc: + memory_dreaming_app.run_dreaming( + request, + authorization=None, + ) + assert exc.value.status_code == 500 + assert exc.value.detail == "failed" diff --git a/test/backend/database/test_memory_dreaming_postgres_integration.py b/test/backend/database/test_memory_dreaming_postgres_integration.py new file mode 100644 index 0000000000..a1251ec2cd --- /dev/null +++ b/test/backend/database/test_memory_dreaming_postgres_integration.py @@ -0,0 +1,89 @@ +"""Opt-in PostgreSQL integration coverage for Dreaming advisory locks.""" + +import os + +import psycopg2 +import pytest + +from database.memory_dreaming_db import advisory_lock_key + +pytestmark = pytest.mark.skipif( + os.getenv("RUN_POSTGRES_INTEGRATION") != "1", + reason="set RUN_POSTGRES_INTEGRATION=1 with local PostgreSQL env", +) + + +def _connect(): + return psycopg2.connect( + host=os.getenv("DREAMING_TEST_POSTGRES_HOST", os.environ["POSTGRES_HOST"]), + port=os.getenv("DREAMING_TEST_POSTGRES_PORT", os.environ["POSTGRES_PORT"]), + user=os.getenv("DREAMING_TEST_POSTGRES_USER", os.environ["POSTGRES_USER"]), + password=os.getenv( + "DREAMING_TEST_POSTGRES_PASSWORD", + os.environ["NEXENT_POSTGRES_PASSWORD"], + ), + dbname=os.getenv("DREAMING_TEST_POSTGRES_DB", os.environ["POSTGRES_DB"]), + ) + + +def _try_lock(connection, lock_key): + with connection.cursor() as cursor: + cursor.execute("SELECT pg_try_advisory_xact_lock(%s)", (lock_key,)) + return cursor.fetchone()[0] + + +def test_ac007_real_postgres_scope_lock_is_non_blocking_and_released(): + same_scope = advisory_lock_key("dreaming-it", "user", "agent") + other_scope = advisory_lock_key("dreaming-it", "user", "other-agent") + + first = _connect() + second = _connect() + try: + assert _try_lock(first, same_scope) is True + assert _try_lock(second, same_scope) is False + assert _try_lock(second, other_scope) is True + + first.rollback() + assert _try_lock(second, same_scope) is True + finally: + first.rollback() + second.rollback() + first.close() + second.close() + + +def test_ac010_real_postgres_audit_schema_matches_orm_contract(): + connection = _connect() + try: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT column_name + FROM information_schema.columns + WHERE table_schema = 'nexent' + AND table_name = 'memory_dreaming_audit_t' + """ + ) + columns = {row[0] for row in cursor.fetchall()} + cursor.execute( + """ + SELECT indexname + FROM pg_indexes + WHERE schemaname = 'nexent' + AND tablename = 'memory_dreaming_audit_t' + """ + ) + indexes = {row[0] for row in cursor.fetchall()} + assert { + "run_id", + "tenant_id", + "user_id", + "agent_id", + "status", + "current_phase", + "result_json", + "error", + } <= columns + assert "idx_memory_dreaming_audit_scope" in indexes + finally: + connection.close() diff --git a/test/backend/database/test_memory_dreaming_schema.py b/test/backend/database/test_memory_dreaming_schema.py new file mode 100644 index 0000000000..cc2500141f --- /dev/null +++ b/test/backend/database/test_memory_dreaming_schema.py @@ -0,0 +1,84 @@ +from pathlib import Path +from datetime import datetime + +from database import memory_retrieval_hit_db +from database.db_models import MemoryDreamingAudit, MemoryRecord, MemoryRetrievalHit +from database.memory_dreaming_db import advisory_lock_key + + +def test_ac010_orm_contract(): + assert MemoryRecord.__tablename__ == "memory_records_t" + assert MemoryRetrievalHit.__tablename__ == "memory_retrieval_hits_t" + columns = MemoryDreamingAudit.__table__.columns + for name in ( + "run_id", + "tenant_id", + "user_id", + "agent_id", + "status", + "current_phase", + "result_json", + "error", + ): + assert name in columns + + +def test_ac007_lock_key_is_stable_and_scope_specific(): + key = advisory_lock_key("tenant", "user", "agent") + assert key == advisory_lock_key("tenant", "user", "agent") + assert key != advisory_lock_key("tenant", "user", "other-agent") + assert -(2**63) <= key < 2**63 + + +def test_ac010_migration_and_fresh_install_match(): + root = Path(__file__).resolve().parents[3] + migration = ( + root / "deploy/sql/migrations/v2.4.0_0723_add_memory_dreaming_audit.sql" + ).read_text() + init_sql = (root / "deploy/sql/init.sql").read_text() + for token in ( + "memory_dreaming_audit_t", + "idx_memory_dreaming_audit_scope", + "result_json", + "promoted_count", + ): + assert token in migration + assert token in init_sql + assert "CREATE TABLE IF NOT EXISTS" in migration + assert "CREATE INDEX IF NOT EXISTS" in migration + + +def test_ac002_dreaming_stats_filter_agent_scope(monkeypatch): + monkeypatch.setattr( + memory_retrieval_hit_db, + "list_hits_for_user", + lambda *_args, **_kwargs: [ + { + "agent_id": "agent-1", + "memory_id": 1, + "day": "2026-07-22", + "query_hash": "q1", + "retrieval_score": 0.75, + "grounded": True, + "occurred_at": datetime(2026, 7, 22, 12), + }, + { + "agent_id": "agent-2", + "memory_id": 2, + "day": "2026-07-22", + "query_hash": "q2", + "retrieval_score": 1.0, + "grounded": True, + "occurred_at": datetime(2026, 7, 22, 13), + }, + ], + ) + rows = memory_retrieval_hit_db.aggregate_dreaming_stats( + "tenant", + "user", + "agent-1", + since=datetime(2026, 7, 20), + ) + assert len(rows) == 1 + assert rows[0]["memory_id"] == 1 + assert rows[0]["total_retrieval_score"] == 0.75 diff --git a/test/backend/services/test_memory_dreaming_scheduler.py b/test/backend/services/test_memory_dreaming_scheduler.py deleted file mode 100644 index 08154fb24d..0000000000 --- a/test/backend/services/test_memory_dreaming_scheduler.py +++ /dev/null @@ -1,193 +0,0 @@ -"""Unit tests for ``backend.services.memory_dreaming_scheduler`` (Phase 2).""" - -import sys -import types -from datetime import datetime -from unittest.mock import MagicMock - -import pytest - - -# Path setup -sys.path.insert( - 0, - __import__("os").path.join(__import__("os").path.dirname(__file__), "../../.."), -) - - -# Stub consts -consts_pkg = types.ModuleType("consts") -consts_pkg.AGENT_SHORT_TERM_HALF_LIFE_DAYS = 14 -consts_pkg.LIGHT_SLEEP_WINDOW_DAYS = 7 -consts_pkg.MIN_PROMOTION_SCORE = 0.72 -consts_pkg.MIN_RECALL_COUNT = 3 -consts_pkg.MIN_UNIQUE_QUERIES = 2 -consts_pkg.RECENCY_HALF_LIFE_DAYS = 14 -consts_mod = types.ModuleType("consts.const") -for name, value in vars(consts_pkg).items(): - if not name.startswith("_"): - setattr(consts_mod, name, value) -sys.modules["consts"] = types.ModuleType("consts") -sys.modules["consts.const"] = consts_mod - - -# Stub database -database_pkg = types.ModuleType("database") -database_pkg.memory_record_db = MagicMock(name="memory_record_db") -database_pkg.memory_retrieval_hit_db = MagicMock(name="memory_retrieval_hit_db") -sys.modules["database"] = database_pkg -sys.modules["backend.database"] = database_pkg - - -# Stub services.memory_record_service -memory_record_service_mod = types.ModuleType("services.memory_record_service") -memory_record_service_mod.MemoryRecordError = type("MemoryRecordError", (Exception,), {}) - - -class _RecordService: - pass - - -memory_record_service_mod.MemoryRecordService = _RecordService -memory_record_service_mod.get_memory_record_service = MagicMock( - name="get_memory_record_service" -) -sys.modules["services.memory_record_service"] = memory_record_service_mod - - -from backend.services import memory_dreaming_scheduler - - -def test_compute_promotion_score_no_signal(): - score = memory_dreaming_scheduler.compute_promotion_score({}) - assert 0.0 <= score <= 1.0 - - -def test_compute_promotion_score_increases_with_recall(): - low = memory_dreaming_scheduler.compute_promotion_score( - {"recall_count": 1, "daily_count": 1, "grounded_count": 1, "light_hits": 0, - "rem_hits": 0, "last_recalled_at": datetime.utcnow(), "concept_tags": [], - "query_hashes": []} - ) - high = memory_dreaming_scheduler.compute_promotion_score( - {"recall_count": 12, "daily_count": 5, "grounded_count": 4, - "light_hits": 3, "rem_hits": 3, - "last_recalled_at": datetime.utcnow(), "concept_tags": ["python"], - "query_hashes": ["a", "b", "c", "d", "e"]} - ) - assert high > low - - -def test_run_light_sleep_aggregates_into_rows(): - memory_dreaming_scheduler.memory_retrieval_hit_db.aggregate_memory_stats.return_value = [ - { - "memory_id": 1, - "hit_count": 5, - "grounded_count": 2, - "days": {"2026-07-13", "2026-07-12"}, - "query_hashes": {"q1", "q2"}, - } - ] - memory_dreaming_scheduler.memory_record_db.update_memory_record.return_value = True - memory_dreaming_scheduler.memory_record_db.apply_dreaming_phase.return_value = True - - touched = memory_dreaming_scheduler.run_light_sleep( - tenant_id="t1", user_id="u1" - ) - - assert touched == 1 - memory_dreaming_scheduler.memory_record_db.update_memory_record.assert_called_once() - memory_dreaming_scheduler.memory_record_db.apply_dreaming_phase.assert_called_once_with( - 1, "t1", phase="light" - ) - - -def test_run_rem_sleep_writes_concept_tags(): - memory_dreaming_scheduler.memory_record_db.list_memory_records.return_value = [ - { - "memory_id": 1, - "tenant_id": "t1", - "user_id": "u1", - "content": "Python Python Java Python Java C++", - "layer": "agent", - "memory_type": "short_term", - "concept_tags": [], - } - ] - memory_dreaming_scheduler.memory_record_db.update_memory_record.return_value = True - memory_dreaming_scheduler.memory_record_db.apply_dreaming_phase.return_value = True - - touched = memory_dreaming_scheduler.run_rem_sleep( - tenant_id="t1", user_id="u1" - ) - - assert touched == 1 - # Update payload should carry the new tags. - update_call = memory_dreaming_scheduler.memory_record_db.update_memory_record.call_args - payload = update_call.args[2] - assert "python" in payload["concept_tags"] - assert "java" in payload["concept_tags"] - - -def test_run_deep_sleep_skips_low_signal(): - memory_dreaming_scheduler.memory_record_db.list_memories_for_dreaming.return_value = [ - { - "memory_id": 1, - "content": "low signal", - "layer": "agent", - "recall_count": 1, - "daily_count": 0, - "grounded_count": 0, - "query_hashes": ["q1"], - "concept_tags": [], - "last_recalled_at": datetime.utcnow(), - "light_hits": 0, - "rem_hits": 0, - } - ] - memory_record_service_mod.get_memory_record_service.return_value.create_memory.return_value = { - "memory_id": 999, - "event": "ADD", - } - - promoted = memory_dreaming_scheduler.run_deep_sleep( - tenant_id="t1", user_id="u1", min_score=0.99 - ) - - assert promoted == [] - memory_record_service_mod.get_memory_record_service.return_value.create_memory.assert_not_called() - - -def test_run_deep_sleep_promotes_high_signal(): - memory_dreaming_scheduler.memory_record_db.list_memories_for_dreaming.return_value = [ - { - "memory_id": 1, - "content": "user prefers dark mode", - "layer": "agent", - "recall_count": 8, - "daily_count": 4, - "grounded_count": 2, - "query_hashes": ["q1", "q2", "q3"], - "concept_tags": ["preference"], - "last_recalled_at": datetime.utcnow(), - "light_hits": 2, - "rem_hits": 1, - "agent_id": "a1", - "conversation_id": "c1", - } - ] - memory_record_service_mod.get_memory_record_service.return_value.create_memory.return_value = { - "memory_id": 999, - "event": "ADD", - } - memory_dreaming_scheduler.memory_record_db.apply_dreaming_phase.return_value = True - - promoted = memory_dreaming_scheduler.run_deep_sleep( - tenant_id="t1", user_id="u1", min_score=0.5 - ) - - assert len(promoted) == 1 - create_kwargs = memory_record_service_mod.get_memory_record_service.return_value.create_memory.call_args.kwargs - assert create_kwargs["layer"] == "user" - assert create_kwargs["memory_type"] == "long_term" - assert create_kwargs["actor"] == "dreaming" \ No newline at end of file diff --git a/test/backend/services/test_memory_dreaming_service.py b/test/backend/services/test_memory_dreaming_service.py new file mode 100644 index 0000000000..9bca862e2a --- /dev/null +++ b/test/backend/services/test_memory_dreaming_service.py @@ -0,0 +1,199 @@ +from contextlib import contextmanager +from datetime import datetime +from unittest.mock import MagicMock + +import pytest + +from nexent.memory.dreaming import DreamingThresholds, select_candidates +from services.memory_dreaming_service import DreamingRunError, MemoryDreamingService + + +@contextmanager +def lock(value): + yield value + + +def test_ac007_lock_busy_skips(monkeypatch): + monkeypatch.setattr( + "services.memory_dreaming_service.memory_dreaming_db.create_audit", + lambda *_: 41, + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_dreaming_db.try_scope_lock", + lambda *_: lock(False), + ) + finish = MagicMock() + monkeypatch.setattr( + "services.memory_dreaming_service.memory_dreaming_db.finish_audit", finish + ) + result = MemoryDreamingService(record_service=MagicMock()).run( + tenant_id="t", user_id="u", agent_id="a" + ) + assert result == {"run_id": 41, "status": "skipped", "reason": "lock_busy"} + finish.assert_called_once() + + +def test_ac001_ac006_full_run_and_idempotency_key(monkeypatch): + record = { + "memory_id": 7, + "tenant_id": "t", + "user_id": "u", + "agent_id": "a", + "content": "Always prefer stable transaction rollback behavior", + "recall_count": 3, + "daily_count": 2, + "grounded_count": 1, + "last_recalled_at": datetime.utcnow().isoformat(), + "query_hashes": ["q1", "q2"], + "recall_days": ["2026-07-22", "2026-07-23"], + "light_hits": 2, + "rem_hits": 2, + "last_light_at": datetime.utcnow().isoformat(), + "last_rem_at": datetime.utcnow().isoformat(), + "concept_tags": [], + } + monkeypatch.setattr( + "services.memory_dreaming_service.memory_dreaming_db.create_audit", + lambda *_: 42, + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_dreaming_db.try_scope_lock", + lambda *_: lock(True), + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_dreaming_db.update_audit", + lambda *_args, **_kwargs: True, + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_dreaming_db.finish_audit", + lambda *_args, **_kwargs: True, + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_retrieval_hit_db.aggregate_dreaming_stats", + lambda *_args, **_kwargs: [ + { + "memory_id": 7, + "hit_count": 4, + "grounded_count": 1, + "days": {"2026-07-22", "2026-07-23"}, + "query_hashes": {"q1", "q2"}, + "total_retrieval_score": 3.8, + "last_recalled_at": datetime.utcnow(), + } + ], + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_record_db.list_memory_records", + lambda *_args, **_kwargs: [record], + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_record_db.find_by_idempotency", + lambda *_args, **_kwargs: None, + ) + update_record = MagicMock(return_value=True) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_record_db.update_memory_record", + update_record, + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_record_db.apply_dreaming_phase", + lambda *_args, **_kwargs: True, + ) + record_service = MagicMock() + record_service.create_memory.return_value = {"event": "ADD"} + result = MemoryDreamingService(record_service=record_service).run( + tenant_id="t", + user_id="u", + agent_id="a", + min_score=0, + min_recall_count=0, + min_unique_queries=0, + ) + assert result["status"] == "completed" + assert result["light_count"] == 1 + assert result["promoted_count"] == 1 + light_payload = update_record.call_args_list[0].args[2] + assert light_payload["recall_count"] == 4 + assert light_payload["daily_count"] == 2 + assert light_payload["grounded_count"] == 1 + assert light_payload["query_hashes"] == ["q1", "q2"] + assert ( + record_service.create_memory.call_args.kwargs["idempotency_key"] == "dreaming:7" + ) + assert record_service.create_memory.call_args.kwargs["layer"] == "user" + + +def test_ac006_already_promoted_candidate_is_not_written_again(monkeypatch): + service = MemoryDreamingService(record_service=MagicMock()) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_record_db.list_memory_records", + lambda *_args, **_kwargs: [ + { + "memory_id": 9, + "tenant_id": "t", + "user_id": "u", + "agent_id": "a", + "content": "Always prefer stable transaction behavior", + "recall_count": 10, + "daily_count": 5, + "grounded_count": 2, + "last_recalled_at": datetime.utcnow().isoformat(), + "query_hashes": ["q1", "q2", "q3"], + "recall_days": ["2026-07-21", "2026-07-22", "2026-07-23"], + "light_hits": 3, + "rem_hits": 3, + "last_light_at": datetime.utcnow().isoformat(), + "last_rem_at": datetime.utcnow().isoformat(), + "concept_tags": ["preference", "transaction"], + } + ], + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_record_db.find_by_idempotency", + lambda *_args, **_kwargs: {"memory_id": 99}, + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_record_db.update_memory_record", + lambda *_args, **_kwargs: True, + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_record_db.apply_dreaming_phase", + lambda *_args, **_kwargs: True, + ) + candidates = service._run_rem("t", "u", "a", {}) + decisions = select_candidates( + candidates, + thresholds=DreamingThresholds( + min_score=0, + min_recall_count=0, + min_unique_queries=0, + ), + ) + result = service._promote(decisions) + assert result[0]["event"] == "DEFER" + assert result[0]["reason"] == "already_promoted" + service.record_service.create_memory.assert_not_called() + + +def test_ac008_failure_is_audited(monkeypatch): + monkeypatch.setattr( + "services.memory_dreaming_service.memory_dreaming_db.create_audit", + lambda *_: 43, + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_dreaming_db.try_scope_lock", + lambda *_: lock(True), + ) + monkeypatch.setattr( + "services.memory_dreaming_service.memory_retrieval_hit_db.aggregate_dreaming_stats", + MagicMock(side_effect=ValueError("bad data")), + ) + finish = MagicMock() + monkeypatch.setattr( + "services.memory_dreaming_service.memory_dreaming_db.finish_audit", finish + ) + service = MemoryDreamingService(record_service=MagicMock()) + with pytest.raises(DreamingRunError): + service.run(tenant_id="t", user_id="u", agent_id="a") + assert finish.call_args.kwargs["status"] == "failed" + assert "ValueError" in finish.call_args.kwargs["error"] diff --git a/test/sdk/memory/test_dreaming.py b/test/sdk/memory/test_dreaming.py new file mode 100644 index 0000000000..3d5f4538a7 --- /dev/null +++ b/test/sdk/memory/test_dreaming.py @@ -0,0 +1,132 @@ +from datetime import datetime, timedelta + +import pytest + +from nexent.memory.dreaming import ( + DreamingCandidate, + DreamingThresholds, + analyze_rem_content, + compute_metrics, + score_candidate, + select_candidates, +) + + +def candidate(**overrides): + values = { + "memory_id": 1, + "tenant_id": "tenant", + "user_id": "user", + "agent_id": "agent", + "content": "The user always prefers PostgreSQL transactions.", + "recall_count": 3, + "daily_count": 2, + "grounded_count": 1, + "total_retrieval_score": 4.2, + "query_hashes": ["q1", "q2"], + "recall_days": ["2026-07-20", "2026-07-22"], + "concept_tags": ["preference", "transaction"], + "light_hits": 2, + "rem_hits": 1, + "last_recalled_at": datetime(2026, 7, 22), + "last_light_at": datetime(2026, 7, 22), + "last_rem_at": datetime(2026, 7, 22), + } + values.update(overrides) + return DreamingCandidate(**values) + + +def test_ac004_openclaw_score_formula(): + score, metrics = score_candidate(candidate(), now=datetime(2026, 7, 23)) + assert score == pytest.approx(0.7268086998271306) + assert metrics.signal_count == 6 + assert metrics.context_diversity == 2 + assert metrics.relevance == pytest.approx(0.7) + assert metrics.consolidation == pytest.approx(0.36544353551179476) + + +@pytest.mark.parametrize( + ("changes", "thresholds", "reason"), + [ + ({"noise": True}, DreamingThresholds(min_score=0), "noise"), + ( + {"already_promoted": True}, + DreamingThresholds(min_score=0), + "already_promoted", + ), + ( + {"total_retrieval_score": 0}, + DreamingThresholds(min_score=0.7), + "score_below_threshold", + ), + ( + {"recall_count": 0, "daily_count": 0, "grounded_count": 0}, + DreamingThresholds(min_score=0, min_recall_count=1), + "recall_below_threshold", + ), + ( + {"query_hashes": [], "recall_days": []}, + DreamingThresholds(min_score=0, min_unique_queries=1), + "diversity_below_threshold", + ), + ], +) +def test_ac005_gates(changes, thresholds, reason): + decisions = select_candidates( + [candidate(**changes)], thresholds=thresholds, now=datetime(2026, 7, 23) + ) + assert decisions[0].promote is False + assert decisions[0].reason == reason + + +def test_ac005_stable_sorting(): + now = datetime(2026, 7, 23) + first = candidate(memory_id=9) + second = candidate(memory_id=2) + decisions = select_candidates( + [first, second], + thresholds=DreamingThresholds( + min_score=0, min_recall_count=0, min_unique_queries=0 + ), + now=now, + ) + assert [item.candidate.memory_id for item in decisions] == [2, 9] + + +def test_ac003_rem_patterns_and_noise(): + tags, noise = analyze_rem_content( + "I always prefer PostgreSQL transaction rollback." + ) + assert {"preference", "persistent", "transaction"} <= set(tags) + assert noise is False + _, noise = analyze_rem_content("Today's temporary TODO for this session") + assert noise is True + + +def test_metrics_boundaries_and_missing_dates(): + metrics = compute_metrics( + candidate( + recall_count=-2, + daily_count=0, + grounded_count=0, + total_retrieval_score=100, + recall_days=[], + last_recalled_at=None, + last_light_at=None, + last_rem_at=None, + ), + now=datetime(2026, 7, 23), + ) + assert metrics.frequency == 0 + assert metrics.relevance == 1 + assert metrics.recency == 0 + assert metrics.consolidation == 0 + assert metrics.phase_boost == 0 + + +def test_future_timestamps_are_clamped_to_full_recency(): + now = datetime(2026, 7, 23) + metrics = compute_metrics( + candidate(last_recalled_at=now + timedelta(days=1)), now=now + ) + assert metrics.recency == 1