diff --git a/.gitignore b/.gitignore index 78993904b..adb702cfb 100644 --- a/.gitignore +++ b/.gitignore @@ -88,3 +88,4 @@ agent_repository_frontend .tokensave .playwright-mcp/ +frontend/test-results/ diff --git a/.sonarcloud.properties b/.sonarcloud.properties new file mode 100644 index 000000000..b20e348a7 --- /dev/null +++ b/.sonarcloud.properties @@ -0,0 +1 @@ +sonar.cpd.exclusions=deploy/sql/init.sql,deploy/sql/migrations/v2.4.0_0723_add_memory_dreaming_audit.sql,deploy/sql/migrations/v2.4.0_0723_add_memory_dreaming_version.sql diff --git a/backend/agents/create_agent_info.py b/backend/agents/create_agent_info.py index 045db63e9..368e6a473 100644 --- a/backend/agents/create_agent_info.py +++ b/backend/agents/create_agent_info.py @@ -101,6 +101,19 @@ def _format_long_term_memory_prompt(search_context: Any, language: str) -> str: return "\n\n".join(sections) +def _get_active_dreaming_version( + tenant_id: str, user_id: str, agent_id: str +) -> Optional[Dict[str, Any]]: + """Keep the Dreaming repository optional in isolated SDK-style tests.""" + try: + from database.memory_dreaming_db import get_active_version + + return get_active_version(tenant_id, user_id, agent_id) + except ModuleNotFoundError: + logger.debug("Dreaming repository is unavailable in this runtime") + return None + + # Safe fallback for context-manager token_threshold when no capacity is known. # Used only when the resolver fails (uncataloged model with no operator-supplied # hard capacity). Sized to cover the typical 32K-context band shared by the @@ -830,6 +843,20 @@ async def create_agent_config( memory_context = build_memory_context( user_id, tenant_id, agent_id, skip_query=not allow_memory_search ) + if allow_memory_search and memory_context.user_config.memory_switch: + active_dreaming_version = _get_active_dreaming_version( + str(tenant_id), str(user_id), str(agent_id) + ) + if active_dreaming_version: + memory_list.append( + { + "memory": active_dreaming_version["published_content"], + "memory_level": "user", + "memory_type": "long_term", + "score": 1.0, + "dreaming_version_id": active_dreaming_version["version_id"], + } + ) # Append active memory tools if memory is enabled if memory_context.user_config.memory_switch: diff --git a/backend/apps/config_app.py b/backend/apps/config_app.py index fe9e9ff5a..38fd58f83 100644 --- a/backend/apps/config_app.py +++ b/backend/apps/config_app.py @@ -43,6 +43,8 @@ from apps.memory_config_app import router as memory_config_router from apps.memory_record_app import router as memory_record_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 @@ -62,6 +64,19 @@ async def sync_default_prompt_template_on_startup(): except Exception as exc: logger.error(f"Failed to sync system default prompt template: {str(exc)}") + +@app.on_event("startup") +async def start_dreaming_scheduler(): + from services.memory_dreaming_scheduler import dreaming_scheduler + await dreaming_scheduler.start() + + +@app.on_event("shutdown") +async def stop_dreaming_scheduler(): + from services.memory_dreaming_scheduler import dreaming_scheduler + await dreaming_scheduler.stop() + + app.include_router(model_manager_router) app.include_router(config_sync_router) app.include_router(agent_router) @@ -111,3 +126,5 @@ async def sync_default_prompt_template_on_startup(): app.include_router(memory_record_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_config_app.py b/backend/apps/memory_config_app.py index f61476656..a80892cd3 100644 --- a/backend/apps/memory_config_app.py +++ b/backend/apps/memory_config_app.py @@ -17,6 +17,7 @@ - DELETE `/memory/config/disable_useragent/{agent_id}`: Remove a disabled user-agent id. """ import logging +from datetime import datetime from typing import Any, Optional from http import HTTPStatus @@ -26,6 +27,7 @@ from consts.const import ( MEMORY_AGENT_SHARE_KEY, MEMORY_SWITCH_KEY, + DREAMING_SWITCH_KEY, BOOLEAN_TRUE_VALUES, ) from consts.model import MemoryAgentShareMode @@ -38,7 +40,9 @@ remove_disabled_useragent_id, set_agent_share, set_memory_switch, + set_dreaming_switch, ) +from database import memory_dreaming_db from services.memory_record_service import ( get_tenant_memory_index_name, is_tenant_embedding_configured, @@ -112,6 +116,9 @@ def set_single_config( raise HTTPException(status_code=HTTPStatus.NOT_ACCEPTABLE, detail="Invalid value for MEMORY_AGENT_SHARE (expected always/ask/never)") ok = set_agent_share(user_id, mode) + elif key == DREAMING_SWITCH_KEY: + enabled = bool(value) if isinstance(value, bool) else str(value).lower() in BOOLEAN_TRUE_VALUES + ok = set_dreaming_switch(user_id, enabled) else: raise HTTPException(status_code=HTTPStatus.NOT_ACCEPTABLE, detail="Unsupported configuration key") @@ -122,6 +129,31 @@ def set_single_config( detail="Failed to update configuration") +@router.post("/config/dreaming") +def set_dreaming_config( + enabled: bool = Body(...), + delete_history: bool = Body(False), + authorization: Optional[str] = Header(None), +): + user_id, tenant_id = get_current_user_id(authorization) + if not set_dreaming_switch(user_id, enabled): + raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail="Failed to update Dreaming") + if not enabled: + schedule = memory_dreaming_db.get_schedule(tenant_id, user_id, "__user__") + if schedule: + memory_dreaming_db.upsert_schedule( + tenant_id, user_id, "__user__", enabled=False, + rule_type=schedule["rule_type"], timezone_name=schedule["timezone"], + start_at=datetime.fromisoformat(schedule["start_at"]), + cron_expr=schedule["cron_expr"], + interval_seconds=schedule["interval_seconds"], + next_fire_at=None, actor_user_id=user_id, + ) + if delete_history: + memory_dreaming_db.delete_user_dreaming_history(tenant_id, user_id) + return {"success": True} + + @router.post("/config/disable_agent") def add_disable_agent( agent_id: str = Body(..., embed=True), diff --git a/backend/apps/memory_dreaming_app.py b/backend/apps/memory_dreaming_app.py new file mode 100644 index 000000000..679f998ef --- /dev/null +++ b/backend/apps/memory_dreaming_app.py @@ -0,0 +1,274 @@ +"""Manual Dreaming run and audit endpoints.""" + +from http import HTTPStatus +from datetime import datetime, timezone +from typing import Annotated, Literal, Optional +from zoneinfo import ZoneInfo + +from fastapi import APIRouter, Header, HTTPException, Query +from pydantic import BaseModel, Field +from pydantic import model_validator +from nexent.scheduler import ScheduleMode, ScheduleRuleType +from services.agent_automation.models import ScheduleTrigger +from services.agent_automation.schedule_engine import ( + compute_next_fire_at, + is_valid_cron_expression, +) + +from consts.const import ( + DREAMING_COMPRESSION_MAX_ATTEMPTS, + DREAMING_LONG_TERM_MAX_CHARS, + DREAMING_SOURCE_LIMIT, +) +from database import memory_dreaming_db +from database.role_permission_db import check_role_permission +from database.user_tenant_db import get_user_tenant_by_user_id +from services.memory_dreaming_service import ( + DreamingConflictError, + DreamingRunError, + get_memory_dreaming_service, +) +from utils.auth_utils import get_current_user_id + +router = APIRouter(prefix="/memory/dreaming", tags=["memory-dreaming"]) +USER_DREAMING_SCOPE = "__user__" + + +class DreamingRunRequest(BaseModel): + target_user_id: Optional[str] = None + + +class DreamingVersionSwitchRequest(BaseModel): + expected_active_version_id: int = Field(..., ge=1) + target_user_id: Optional[str] = None + + +class DreamingScheduleRequest(BaseModel): + enabled: bool + rule_type: Literal["CRON", "INTERVAL"] = "CRON" + timezone: str = "Asia/Shanghai" + start_at: Optional[datetime] = None + cron_expr: Optional[str] = None + interval_seconds: Optional[int] = Field(default=None, ge=3600) + target_user_id: Optional[str] = None + + @model_validator(mode="after") + def validate_schedule(self): + try: + ZoneInfo(self.timezone) + except Exception as exc: + raise ValueError(f"Invalid timezone: {self.timezone}") from exc + if self.rule_type == "CRON": + if not is_valid_cron_expression(self.cron_expr or ""): + raise ValueError("A valid five-field cron_expr is required") + if self.interval_seconds is not None: + raise ValueError("CRON schedule cannot include interval_seconds") + else: + if self.interval_seconds is None: + raise ValueError("interval_seconds is required") + if self.cron_expr is not None: + raise ValueError("INTERVAL schedule cannot include cron_expr") + return self + + +def _resolve_target_user( + authorization: Optional[str], + target_user_id: Optional[str], + *, + tenant_capability: str, +) -> tuple[str, str]: + caller_user_id, tenant_id = get_current_user_id(authorization) + if not target_user_id or target_user_id == caller_user_id: + return caller_user_id, tenant_id + caller = get_user_tenant_by_user_id(caller_user_id) or {} + caller_role = str(caller.get("user_role") or "").upper() + if not check_role_permission( + caller_role, + permission_category="RESOURCE", + permission_type="DREAMING", + permission_subtype=tenant_capability, + ): + raise HTTPException( + status_code=HTTPStatus.NOT_FOUND, detail="Resource not found" + ) + target = get_user_tenant_by_user_id(target_user_id) or {} + if target.get("tenant_id") != tenant_id: + raise HTTPException( + status_code=HTTPStatus.NOT_FOUND, detail="Resource not found" + ) + return target_user_id, tenant_id + + + +@router.get("/parameters") +def get_dreaming_parameters( + authorization: Annotated[Optional[str], Header()] = None, +): + """Expose effective read-only build parameters for an authenticated user.""" + get_current_user_id(authorization) + return { + "source_limit": DREAMING_SOURCE_LIMIT, + "long_term_max_chars": DREAMING_LONG_TERM_MAX_CHARS, + "compression_max_attempts": DREAMING_COMPRESSION_MAX_ATTEMPTS, + } + + +@router.get("/schedule") +def get_dreaming_schedule( + agent_id: Annotated[Optional[str], Query()] = None, + authorization: Annotated[Optional[str], Header()] = None, + target_user_id: Annotated[Optional[str], Query()] = None, +): + user_id, tenant_id = _resolve_target_user( + authorization, target_user_id, tenant_capability="VIEW_TENANT" + ) + schedule = memory_dreaming_db.get_schedule(tenant_id, user_id, USER_DREAMING_SCOPE) + return schedule or { + "agent_id": USER_DREAMING_SCOPE, + "enabled": False, + "rule_type": "CRON", + "timezone": "Asia/Shanghai", + "start_at": None, + "cron_expr": "0 3 * * *", + "interval_seconds": None, + "next_fire_at": None, + "last_fire_at": None, + "fire_count": 0, + } + + +@router.put("/schedule") +def put_dreaming_schedule( + payload: DreamingScheduleRequest, + authorization: Annotated[Optional[str], Header()] = None, +): + user_id, tenant_id = _resolve_target_user( + authorization, payload.target_user_id, tenant_capability="EDIT_TENANT" + ) + actor_user_id, _ = get_current_user_id(authorization) + now = datetime.now(timezone.utc) + start_at = payload.start_at or now + spec = ScheduleTrigger( + mode=ScheduleMode.RECURRING, + rule_type=ScheduleRuleType(payload.rule_type), + timezone=payload.timezone, + start_at=start_at, + cron_expr=payload.cron_expr, + interval_seconds=payload.interval_seconds, + ) + next_fire_at = compute_next_fire_at(spec, now, 0) if payload.enabled else None + return memory_dreaming_db.upsert_schedule( + tenant_id, + user_id, + USER_DREAMING_SCOPE, + enabled=payload.enabled, + rule_type=payload.rule_type, + timezone_name=payload.timezone, + start_at=( + start_at.replace(tzinfo=ZoneInfo(payload.timezone)) + if start_at.tzinfo is None + else start_at.astimezone(ZoneInfo(payload.timezone)) + ).replace(tzinfo=None), + cron_expr=payload.cron_expr, + interval_seconds=payload.interval_seconds, + next_fire_at=( + next_fire_at.astimezone(timezone.utc).replace(tzinfo=None) + if next_fire_at + else None + ), + actor_user_id=actor_user_id, + ) + + +@router.post("/run", status_code=HTTPStatus.ACCEPTED) +def run_dreaming( + payload: DreamingRunRequest, + authorization: Annotated[Optional[str], Header()] = None, +): + user_id, tenant_id = _resolve_target_user( + authorization, + payload.target_user_id, + tenant_capability="EDIT_TENANT", + ) + try: + run_id = memory_dreaming_db.create_audit( + tenant_id, + user_id, + USER_DREAMING_SCOPE, + trigger_source="manual", + status="queued", + ) + return {"run_id": run_id, "status": "queued"} + 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, + target_user_id: Annotated[Optional[str], Query()] = None, +): + user_id, tenant_id = _resolve_target_user( + authorization, + target_user_id, + tenant_capability="VIEW_TENANT", + ) + return get_memory_dreaming_service().list_audits( + tenant_id, + user_id, + agent_id=USER_DREAMING_SCOPE, + run_id=run_id, + limit=limit, + ) + + +@router.get("/versions") +def list_dreaming_versions( + agent_id: Annotated[Optional[str], Query()] = None, + authorization: Annotated[Optional[str], Header()] = None, + limit: Annotated[int, Query(ge=1, le=500)] = 100, + target_user_id: Annotated[Optional[str], Query()] = None, +): + user_id, tenant_id = _resolve_target_user( + authorization, + target_user_id, + tenant_capability="VIEW_TENANT", + ) + return get_memory_dreaming_service().list_versions( + tenant_id, user_id, agent_id=USER_DREAMING_SCOPE, limit=limit + ) + + +@router.post("/versions/{version_id}/activate") +def activate_dreaming_version( + version_id: int, + payload: DreamingVersionSwitchRequest, + authorization: Annotated[Optional[str], Header()] = None, +): + user_id, tenant_id = _resolve_target_user( + authorization, + payload.target_user_id, + tenant_capability="EDIT_TENANT", + ) + actor_user_id, _ = get_current_user_id(authorization) + try: + version = get_memory_dreaming_service().activate_version( + tenant_id, + user_id, + agent_id=USER_DREAMING_SCOPE, + version_id=version_id, + actor_user_id=actor_user_id, + expected_active_version_id=payload.expected_active_version_id, + ) + except DreamingConflictError as exc: + raise HTTPException(status_code=HTTPStatus.CONFLICT, detail=str(exc)) from exc + if version is None: + raise HTTPException( + status_code=HTTPStatus.NOT_FOUND, detail="Version not found" + ) + return version diff --git a/backend/consts/const.py b/backend/consts/const.py index a1b84ad1e..0a103380a 100644 --- a/backend/consts/const.py +++ b/backend/consts/const.py @@ -4,7 +4,10 @@ from dotenv import load_dotenv # Load environment variables -load_dotenv(override=True) +# Explicitly sourced deployment variables take precedence over a nearby +# developer .env file. This is required for tmux/K8s-local verification and +# avoids silently replacing operator-provided service addresses. +load_dotenv(override=False) # TODO: Analyze every variable if this is used # Test voice file path (WAV format for volcengine STT) @@ -318,10 +321,12 @@ class VectorDatabaseType(str, Enum): # Memory Feature MEMORY_SWITCH_KEY = "MEMORY_SWITCH" +DREAMING_SWITCH_KEY = "DREAMING_SWITCH" MEMORY_AGENT_SHARE_KEY = "MEMORY_AGENT_SHARE" DISABLE_AGENT_ID_KEY = "DISABLE_AGENT_ID" DISABLE_USERAGENT_ID_KEY = "DISABLE_USERAGENT_ID" DEFAULT_MEMORY_SWITCH_KEY = "Y" +DEFAULT_DREAMING_SWITCH_KEY = "Y" DEFAULT_MEMORY_AGENT_SHARE_KEY = "always" # Boolean value representations for configuration parsing BOOLEAN_TRUE_VALUES = {"true", "1", "y", "yes", "on"} @@ -352,10 +357,17 @@ class VectorDatabaseType(str, Enum): MIN_PROMOTION_SCORE = float(os.getenv("MIN_PROMOTION_SCORE", "0.72")) MIN_RECALL_COUNT = int(os.getenv("MIN_RECALL_COUNT", "3")) MIN_UNIQUE_QUERIES = int(os.getenv("MIN_UNIQUE_QUERIES", "2")) -# Scheduling/cron constants are intentionally not defined here: the -# background Dreaming scheduler is not part of Phase 2 (an agent-driven -# timer will be added in a later phase, at which point the cron expression -# and heartbeat can be reintroduced). +DREAMING_SOURCE_LIMIT = int(os.getenv("DREAMING_SOURCE_LIMIT", "10")) +DREAMING_LONG_TERM_MAX_CHARS = int( + os.getenv("DREAMING_LONG_TERM_MAX_CHARS", "10000") +) +DREAMING_COMPRESSION_MAX_ATTEMPTS = int( + os.getenv("DREAMING_COMPRESSION_MAX_ATTEMPTS", "2") +) +DREAMING_SCHEDULER_POLL_SECONDS = float(os.getenv("DREAMING_SCHEDULER_POLL_SECONDS", "5.0")) +DREAMING_SCHEDULER_LEASE_SECONDS = float(os.getenv("DREAMING_SCHEDULER_LEASE_SECONDS", "120.0")) +DREAMING_SCHEDULER_MAX_CONCURRENCY = int(os.getenv("DREAMING_SCHEDULER_MAX_CONCURRENCY", "1")) +DREAMING_SCHEDULER_ENABLED = os.getenv("DREAMING_SCHEDULER_ENABLED", "true").lower() in ("true", "1", "yes") # External provider retry / timeout PROVIDER_RETRY_MAX_ATTEMPTS = int(os.getenv("PROVIDER_RETRY_MAX_ATTEMPTS", "3")) diff --git a/backend/data_process/app.py b/backend/data_process/app.py index e70403c36..875d015e9 100644 --- a/backend/data_process/app.py +++ b/backend/data_process/app.py @@ -48,7 +48,7 @@ task_routes={ f'{import_path}.process': {'queue': 'process_q'}, f'{import_path}.forward': {'queue': 'forward_q'}, - f'{import_path}.process_and_forward': {'queue': 'process_q'} + f'{import_path}.process_and_forward': {'queue': 'process_q'}, }, task_serializer='json', accept_content=['json'], diff --git a/backend/database/db_models.py b/backend/database/db_models.py index aec452479..609f19488 100644 --- a/backend/database/db_models.py +++ b/backend/database/db_models.py @@ -937,6 +937,167 @@ class MemoryRetrievalHit(TableBase): doc="Soft delete flag (N = active, Y = deleted).") +class MemoryDreamingAudit(TableBase): + """One durable audit row per manual or scheduled 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) + lock_owner = Column(String(100), nullable=True) + lock_until = Column(TIMESTAMP(timezone=False), nullable=True) + + +class MemoryDreamingSchedule(TableBase): + """Persistent automatic Dreaming schedule for one user/agent scope.""" + + __tablename__ = "memory_dreaming_schedule_t" + __table_args__ = ( + Index( + "uq_memory_dreaming_schedule_scope", + "tenant_id", + "user_id", + "agent_id", + unique=True, + ), + Index( + "idx_memory_dreaming_schedule_due", + "enabled", + "next_fire_at", + ), + {"schema": SCHEMA}, + ) + + schedule_id = Column( + BigInteger, + Sequence("memory_dreaming_schedule_t_schedule_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) + enabled = Column(Boolean, nullable=False, default=False) + rule_type = Column(String(20), nullable=False, default="CRON") + timezone = Column(String(100), nullable=False, default="Asia/Shanghai") + start_at = Column(TIMESTAMP(timezone=False), nullable=False) + cron_expr = Column(String(100)) + interval_seconds = Column(Integer) + next_fire_at = Column(TIMESTAMP(timezone=False)) + last_fire_at = Column(TIMESTAMP(timezone=False)) + fire_count = Column(Integer, nullable=False, default=0) + + +class MemoryDreamingVersion(TableBase): + """Immutable long-term memory artifact produced by one Dreaming run.""" + + __tablename__ = "memory_dreaming_version_t" + __table_args__ = ( + Index( + "idx_memory_dreaming_version_scope", + "tenant_id", + "user_id", + "agent_id", + "version_no", + unique=True, + ), + Index( + "uq_memory_dreaming_version_active_scope", + "tenant_id", + "user_id", + "agent_id", + unique=True, + postgresql_where=text("is_active AND delete_flag = 'N'"), + ), + Index("uq_memory_dreaming_version_run", "run_id", unique=True), + {"schema": SCHEMA}, + ) + + version_id = Column( + BigInteger, + Sequence("memory_dreaming_version_t_version_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) + version_no = Column(Integer, nullable=False) + parent_version_id = Column(BigInteger) + run_id = Column(BigInteger, nullable=False) + is_active = Column(Boolean, nullable=False, default=False) + raw_content = Column(Text, nullable=False) + published_content = Column(Text, nullable=False) + published_units = Column(JSONB, nullable=False, default=list) + source_evidence_ids = Column(JSONB, nullable=False, default=list) + config_snapshot = Column(JSONB, nullable=False, default=dict) + raw_char_count = Column(Integer, nullable=False) + published_char_count = Column(Integer, nullable=False) + compression_status = Column(String(30), nullable=False) + compression_attempts = Column(Integer, nullable=False, default=0) + compression_audit = Column(JSONB, nullable=False, default=list) + omitted_evidence_ids = Column(JSONB, nullable=False, default=list) + mechanical_truncation = Column(Boolean, nullable=False, default=False) + + +class MemoryDreamingActivationAudit(TableBase): + """Append-only audit for active-version pointer changes.""" + + __tablename__ = "memory_dreaming_activation_audit_t" + __table_args__ = ( + Index( + "idx_memory_dreaming_activation_scope", + "tenant_id", + "user_id", + "agent_id", + "create_time", + ), + {"schema": SCHEMA}, + ) + + activation_id = Column( + BigInteger, + Sequence("memory_dreaming_activation_audit_t_activation_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) + actor_user_id = Column(String(100), nullable=False) + from_version_id = Column(BigInteger) + to_version_id = Column(BigInteger, nullable=False) + reason = Column(String(100), nullable=False, default="user_switch") + + class McpRecord(TableBase): """ MCP (Model Context Protocol) records table diff --git a/backend/database/memory_dreaming_db.py b/backend/database/memory_dreaming_db.py new file mode 100644 index 000000000..ef859cbe9 --- /dev/null +++ b/backend/database/memory_dreaming_db.py @@ -0,0 +1,637 @@ +"""Persistence, scheduling, and PostgreSQL locking for 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 func, text + +from .client import get_db_session +from .db_models import ( + MemoryDreamingActivationAudit, + MemoryDreamingAudit, + MemoryDreamingSchedule, + MemoryDreamingVersion, + MemoryRecord, +) +from nexent.scheduler import ScheduleMode, ScheduleRuleType +from services.agent_automation.models import ScheduleTrigger +from services.agent_automation.schedule_engine import compute_next_fire_at + + +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, + *, + trigger_source: str = "manual", + status: str = "running", +) -> int: + with get_db_session() as session: + row = MemoryDreamingAudit( + tenant_id=tenant_id, + user_id=user_id, + agent_id=agent_id, + trigger_source=trigger_source, + status=status, + current_phase=None if status == "queued" else "light", + ) + session.add(row) + session.commit() + return int(row.run_id) + + +def _schedule_to_dict(row: MemoryDreamingSchedule) -> Dict[str, Any]: + return { + "schedule_id": row.schedule_id, + "agent_id": row.agent_id, + "enabled": row.enabled, + "rule_type": row.rule_type, + "timezone": row.timezone, + "start_at": row.start_at.isoformat() if row.start_at else None, + "cron_expr": row.cron_expr, + "interval_seconds": row.interval_seconds, + "next_fire_at": ( + row.next_fire_at.replace(tzinfo=timezone.utc).isoformat() + if row.next_fire_at + else None + ), + "last_fire_at": ( + row.last_fire_at.replace(tzinfo=timezone.utc).isoformat() + if row.last_fire_at + else None + ), + "fire_count": row.fire_count, + } + + +def get_schedule( + tenant_id: str, user_id: str, agent_id: str +) -> Optional[Dict[str, Any]]: + with get_db_session() as session: + row = ( + session.query(MemoryDreamingSchedule) + .filter( + MemoryDreamingSchedule.tenant_id == tenant_id, + MemoryDreamingSchedule.user_id == user_id, + MemoryDreamingSchedule.agent_id == agent_id, + MemoryDreamingSchedule.delete_flag == "N", + ) + .first() + ) + return _schedule_to_dict(row) if row else None + + +def upsert_schedule( + tenant_id: str, + user_id: str, + agent_id: str, + *, + enabled: bool, + rule_type: str, + timezone_name: str, + start_at: datetime, + cron_expr: Optional[str], + interval_seconds: Optional[int], + next_fire_at: Optional[datetime], + actor_user_id: str, +) -> Dict[str, Any]: + with get_db_session() as session: + row = ( + session.query(MemoryDreamingSchedule) + .filter( + MemoryDreamingSchedule.tenant_id == tenant_id, + MemoryDreamingSchedule.user_id == user_id, + MemoryDreamingSchedule.agent_id == agent_id, + MemoryDreamingSchedule.delete_flag == "N", + ) + .with_for_update() + .first() + ) + if row is None: + row = MemoryDreamingSchedule( + tenant_id=tenant_id, + user_id=user_id, + agent_id=agent_id, + created_by=actor_user_id, + ) + session.add(row) + row.enabled = enabled + row.rule_type = rule_type + row.timezone = timezone_name + row.start_at = start_at + row.cron_expr = cron_expr + row.interval_seconds = interval_seconds + row.next_fire_at = next_fire_at if enabled else None + row.updated_by = actor_user_id + session.flush() + return _schedule_to_dict(row) + + +def _next_schedule_fire(row: MemoryDreamingSchedule, after: datetime) -> Optional[datetime]: + spec = ScheduleTrigger( + mode=ScheduleMode.RECURRING, + rule_type=ScheduleRuleType(row.rule_type), + timezone=row.timezone, + start_at=row.start_at, + cron_expr=row.cron_expr, + interval_seconds=row.interval_seconds, + ) + value = compute_next_fire_at( + spec, after.replace(tzinfo=timezone.utc), row.fire_count + 1 + ) + return value.astimezone(timezone.utc).replace(tzinfo=None) if value else None + + +def materialize_due_schedules(limit: int = 10) -> int: + """Atomically enqueue due schedules and advance them exactly once.""" + now = datetime.now(timezone.utc).replace(tzinfo=None) + created = 0 + with get_db_session() as session: + rows = ( + session.query(MemoryDreamingSchedule) + .filter( + MemoryDreamingSchedule.enabled.is_(True), + MemoryDreamingSchedule.next_fire_at <= now, + MemoryDreamingSchedule.delete_flag == "N", + ) + .order_by(MemoryDreamingSchedule.next_fire_at.asc()) + .limit(limit) + .with_for_update(skip_locked=True) + .all() + ) + for row in rows: + scheduled_fire_at = row.next_fire_at + active = ( + session.query(MemoryDreamingAudit.run_id) + .filter( + MemoryDreamingAudit.tenant_id == row.tenant_id, + MemoryDreamingAudit.user_id == row.user_id, + MemoryDreamingAudit.agent_id == row.agent_id, + MemoryDreamingAudit.status.in_(("queued", "running")), + MemoryDreamingAudit.delete_flag == "N", + ) + .first() + ) + if active is None: + session.add( + MemoryDreamingAudit( + tenant_id=row.tenant_id, + user_id=row.user_id, + agent_id=row.agent_id, + trigger_source="schedule", + status="queued", + ) + ) + created += 1 + row.last_fire_at = scheduled_fire_at + row.fire_count += 1 + row.next_fire_at = _next_schedule_fire(row, now) + return created + + +def get_active_version( + tenant_id: str, user_id: str, agent_id: str +) -> Optional[Dict[str, Any]]: + with get_db_session() as session: + row = ( + session.query(MemoryDreamingVersion) + .filter( + MemoryDreamingVersion.tenant_id == tenant_id, + MemoryDreamingVersion.user_id == user_id, + MemoryDreamingVersion.agent_id.in_((agent_id, "__user__")), + MemoryDreamingVersion.is_active.is_(True), + MemoryDreamingVersion.delete_flag == "N", + ) + .order_by((MemoryDreamingVersion.agent_id == "__user__").desc()) + .first() + ) + return _version_to_dict(row) if row else None + + +def create_and_activate_version( + *, + tenant_id: str, + user_id: str, + agent_id: str, + run_id: int, + parent_version_id: Optional[int], + raw_content: str, + published_content: str, + published_units: List[Dict[str, Any]], + source_evidence_ids: List[str], + config_snapshot: Dict[str, Any], + raw_char_count: int, + published_char_count: int, + compression_status: str, + compression_attempts: int, + omitted_evidence_ids: List[str], + mechanical_truncation: bool, + compression_audit: List[Dict[str, Any]], +) -> Dict[str, Any]: + """Atomically append and activate a version for one locked scope.""" + with get_db_session() as session: + existing = ( + session.query(MemoryDreamingVersion) + .filter( + MemoryDreamingVersion.run_id == run_id, + MemoryDreamingVersion.tenant_id == tenant_id, + MemoryDreamingVersion.user_id == user_id, + MemoryDreamingVersion.agent_id == agent_id, + MemoryDreamingVersion.delete_flag == "N", + ) + .first() + ) + if existing is not None: + return _version_to_dict(existing) + scope = ( + MemoryDreamingVersion.tenant_id == tenant_id, + MemoryDreamingVersion.user_id == user_id, + MemoryDreamingVersion.agent_id == agent_id, + MemoryDreamingVersion.delete_flag == "N", + ) + next_version = ( + session.query(func.coalesce(func.max(MemoryDreamingVersion.version_no), 0)) + .filter(*scope) + .scalar() + + 1 + ) + session.query(MemoryDreamingVersion).filter( + *scope, MemoryDreamingVersion.is_active.is_(True) + ).update({"is_active": False}, synchronize_session=False) + row = MemoryDreamingVersion( + tenant_id=tenant_id, + user_id=user_id, + agent_id=agent_id, + version_no=next_version, + parent_version_id=parent_version_id, + run_id=run_id, + is_active=True, + raw_content=raw_content, + published_content=published_content, + published_units=published_units, + source_evidence_ids=source_evidence_ids, + config_snapshot=config_snapshot, + raw_char_count=raw_char_count, + published_char_count=published_char_count, + compression_status=compression_status, + compression_attempts=compression_attempts, + omitted_evidence_ids=omitted_evidence_ids, + mechanical_truncation=mechanical_truncation, + compression_audit=compression_audit, + created_by="dreaming", + ) + session.add(row) + session.commit() + session.refresh(row) + return _version_to_dict(row) + + +def list_versions( + tenant_id: str, + user_id: str, + *, + agent_id: str, + limit: int = 100, +) -> List[Dict[str, Any]]: + with get_db_session() as session: + rows = ( + session.query(MemoryDreamingVersion) + .filter( + MemoryDreamingVersion.tenant_id == tenant_id, + MemoryDreamingVersion.user_id == user_id, + MemoryDreamingVersion.agent_id == agent_id, + MemoryDreamingVersion.delete_flag == "N", + ) + .order_by(MemoryDreamingVersion.version_no.desc()) + .limit(limit) + .all() + ) + return [_version_to_dict(row) for row in rows] + + +def activate_version( + tenant_id: str, + user_id: str, + agent_id: str, + version_id: int, + *, + actor_user_id: Optional[str] = None, +) -> Optional[Dict[str, Any]]: + """Switch the active pointer without modifying immutable version content.""" + with get_db_session() as session: + rows = ( + session.query(MemoryDreamingVersion) + .filter( + MemoryDreamingVersion.tenant_id == tenant_id, + MemoryDreamingVersion.user_id == user_id, + MemoryDreamingVersion.agent_id == agent_id, + MemoryDreamingVersion.delete_flag == "N", + ) + .all() + ) + target = next((row for row in rows if row.version_id == version_id), None) + if target is None: + return None + current = next((row for row in rows if row.is_active), None) + if current is not None and current.version_id == version_id: + return _version_to_dict(target) + actor = actor_user_id or user_id + session.query(MemoryDreamingVersion).filter( + MemoryDreamingVersion.tenant_id == tenant_id, + MemoryDreamingVersion.user_id == user_id, + MemoryDreamingVersion.agent_id == agent_id, + MemoryDreamingVersion.delete_flag == "N", + MemoryDreamingVersion.is_active.is_(True), + ).update( + {"is_active": False, "updated_by": actor}, + synchronize_session=False, + ) + session.flush() + target.is_active = True + target.updated_by = actor + session.add( + MemoryDreamingActivationAudit( + tenant_id=tenant_id, + user_id=user_id, + agent_id=agent_id, + actor_user_id=actor, + from_version_id=current.version_id if current else None, + to_version_id=version_id, + reason="user_switch", + created_by=actor, + ) + ) + session.commit() + session.refresh(target) + return _version_to_dict(target) + + +def _version_to_dict(row: MemoryDreamingVersion) -> Dict[str, Any]: + return { + "version_id": row.version_id, + "tenant_id": row.tenant_id, + "user_id": row.user_id, + "agent_id": row.agent_id, + "version_no": row.version_no, + "parent_version_id": row.parent_version_id, + "run_id": row.run_id, + "is_active": row.is_active, + "raw_content": row.raw_content, + "published_content": row.published_content, + "published_units": row.published_units or [], + "source_evidence_ids": row.source_evidence_ids or [], + "config_snapshot": row.config_snapshot or {}, + "raw_char_count": row.raw_char_count, + "published_char_count": row.published_char_count, + "compression_status": row.compression_status, + "compression_attempts": row.compression_attempts, + "omitted_evidence_ids": row.omitted_evidence_ids or [], + "mechanical_truncation": row.mechanical_truncation, + "compression_audit": row.compression_audit or [], + "created_at": row.create_time.isoformat() if row.create_time else None, + } + + +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 + ] + + +# --------------------------------------------------------------------------- +# Worker lease management +# --------------------------------------------------------------------------- + + +def claim_queued(owner_id: str, lease_seconds: float) -> Optional[Dict[str, Any]]: + """Atomically claim the oldest queued audit row and set a lease. + + Uses FOR UPDATE SKIP LOCKED so concurrent workers never block each other. + Returns the payload the executor needs (run_id, tenant_id, user_id, + agent_id, trigger_source) or None when no row is available. + """ + sql = text(""" + WITH candidate AS ( + SELECT run_id + FROM nexent.memory_dreaming_audit_t + WHERE status = 'queued' + AND delete_flag = 'N' + ORDER BY started_at ASC + LIMIT 1 + FOR UPDATE SKIP LOCKED + ) + UPDATE nexent.memory_dreaming_audit_t AS audit + SET lock_owner = :owner_id, + lock_until = now() + (:lease_seconds * interval '1 second'), + status = 'running', + current_phase = 'light', + update_time = now() + FROM candidate + WHERE audit.run_id = candidate.run_id + RETURNING audit.run_id, + audit.tenant_id, + audit.user_id, + audit.agent_id, + audit.trigger_source + """) + with get_db_session() as session: + row = session.execute(sql, { + "owner_id": owner_id, + "lease_seconds": lease_seconds, + }).fetchone() + if row is None: + return None + return dict(row._mapping) + + +def renew_lease(run_id: int, owner_id: str, lease_seconds: float) -> bool: + """Extend the lease only when the caller still owns it and it has not expired.""" + sql = text(""" + UPDATE nexent.memory_dreaming_audit_t + SET lock_until = now() + (:lease_seconds * interval '1 second'), + update_time = now() + WHERE run_id = :run_id + AND lock_owner = :owner_id + AND lock_until > now() + AND delete_flag = 'N' + RETURNING run_id + """) + with get_db_session() as session: + renewed = session.execute(sql, { + "run_id": run_id, + "owner_id": owner_id, + "lease_seconds": lease_seconds, + }).scalar_one_or_none() + return renewed is not None + + +def release_lease(run_id: int, owner_id: str) -> bool: + """Clear the lease fields only when the caller owns the lock.""" + sql = text(""" + UPDATE nexent.memory_dreaming_audit_t + SET lock_owner = NULL, + lock_until = NULL, + update_time = now() + WHERE run_id = :run_id + AND lock_owner = :owner_id + AND delete_flag = 'N' + RETURNING run_id + """) + with get_db_session() as session: + released = session.execute(sql, { + "run_id": run_id, + "owner_id": owner_id, + }).scalar_one_or_none() + return released is not None + + +def recover_stale() -> int: + """Reap runs whose lease expired without completion. + + Marks them as failed and clears lock fields so they can be retried. + Safe to call on every worker startup. + """ + sql = text(""" + UPDATE nexent.memory_dreaming_audit_t + SET status = 'failed', + error = 'Worker lost — reaped by startup recovery', + lock_owner = NULL, + lock_until = NULL, + finished_at = now(), + update_time = now() + WHERE status = 'running' + AND lock_until < now() + AND delete_flag = 'N' + """) + with get_db_session() as session: + result = session.execute(sql) + return result.rowcount or 0 + + +def delete_user_dreaming_history(tenant_id: str, user_id: str) -> None: + with get_db_session() as session: + for model in ( + MemoryDreamingSchedule, + MemoryDreamingAudit, + MemoryDreamingVersion, + MemoryDreamingActivationAudit, + ): + session.query(model).filter( + model.tenant_id == tenant_id, + model.user_id == user_id, + model.delete_flag == "N", + ).update( + {"delete_flag": "Y", "updated_by": user_id}, + synchronize_session=False, + ) + session.query(MemoryRecord).filter( + MemoryRecord.tenant_id == tenant_id, + MemoryRecord.user_id == user_id, + MemoryRecord.created_by == "dreaming", + MemoryRecord.delete_flag == "N", + ).update( + {"delete_flag": "Y", "updated_by": user_id}, + synchronize_session=False, + ) diff --git a/backend/database/memory_record_db.py b/backend/database/memory_record_db.py index d81f684b2..b92670b04 100644 --- a/backend/database/memory_record_db.py +++ b/backend/database/memory_record_db.py @@ -329,7 +329,9 @@ def list_memory_records( query = query.filter(MemoryRecord.delete_flag == "N") query = query.order_by(MemoryRecord.update_time.desc()) - query = query.limit(limit).offset(offset) + if limit is not None: + query = query.limit(limit) + query = query.offset(offset) result = [] for ( record, diff --git a/backend/database/memory_retrieval_hit_db.py b/backend/database/memory_retrieval_hit_db.py index 9028ad41b..c140a0542 100644 --- a/backend/database/memory_retrieval_hit_db.py +++ b/backend/database/memory_retrieval_hit_db.py @@ -9,7 +9,7 @@ import logging from datetime import datetime -from typing import Any, Dict, Iterable, List, Optional, Sequence +from typing import Any, Dict, Iterable, List, Optional from sqlalchemy import Integer, func @@ -186,6 +186,50 @@ def aggregate_memory_stats( return out +def aggregate_dreaming_stats( + tenant_id: str, + user_id: str, + agent_id: Optional[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 ( + (agent_id is not None and 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 +267,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_config_service.py b/backend/services/memory_config_service.py index f2eda95fb..88c8f7354 100644 --- a/backend/services/memory_config_service.py +++ b/backend/services/memory_config_service.py @@ -3,10 +3,12 @@ from consts.const import ( MEMORY_SWITCH_KEY, + DREAMING_SWITCH_KEY, MEMORY_AGENT_SHARE_KEY, DISABLE_AGENT_ID_KEY, DISABLE_USERAGENT_ID_KEY, DEFAULT_MEMORY_SWITCH_KEY, + DEFAULT_DREAMING_SWITCH_KEY, DEFAULT_MEMORY_AGENT_SHARE_KEY, ) from consts.model import MemoryAgentShareMode @@ -58,6 +60,8 @@ def get_user_configs(user_id: str) -> Dict[str, Union[str, List[str]]]: aggregated[MEMORY_SWITCH_KEY] = DEFAULT_MEMORY_SWITCH_KEY if MEMORY_AGENT_SHARE_KEY not in aggregated: aggregated[MEMORY_AGENT_SHARE_KEY] = DEFAULT_MEMORY_AGENT_SHARE_KEY + if DREAMING_SWITCH_KEY not in aggregated: + aggregated[DREAMING_SWITCH_KEY] = DEFAULT_DREAMING_SWITCH_KEY return aggregated @@ -157,6 +161,10 @@ def set_memory_switch(user_id: str, enabled: bool) -> bool: return _update_single_config(user_id, MEMORY_SWITCH_KEY, "Y" if enabled else "N") +def set_dreaming_switch(user_id: str, enabled: bool) -> bool: + return _update_single_config(user_id, DREAMING_SWITCH_KEY, "Y" if enabled else "N") + + # Agent share (single string among always/ask/never) def get_agent_share(user_id: str) -> MemoryAgentShareMode: configs = get_user_configs(user_id) diff --git a/backend/services/memory_dreaming_compressor.py b/backend/services/memory_dreaming_compressor.py new file mode 100644 index 000000000..b0f598b01 --- /dev/null +++ b/backend/services/memory_dreaming_compressor.py @@ -0,0 +1,555 @@ +"""Tenant-model semantic compressor for bounded Dreaming versions.""" + +from __future__ import annotations + +import json +import re + +from consts.const import MODEL_CONFIG_MAPPING +from nexent.core.models import OpenAIModel +from nexent.memory.dreaming import ( + DreamingCompressionOutput, + DreamingCompressionRequest, +) +from nexent.monitor import ( + AgentRunMetadata, + agent_monitoring_context, + set_monitoring_operation, +) +from utils.config_utils import get_model_name_from_config, tenant_config_manager + + +def _strip_json_fence(content: object) -> str: + """Remove one optional Markdown JSON fence without regex backtracking.""" + raw = str(content or "").strip() + if not raw.startswith("```"): + return raw + first_line_end = raw.find("\n") + if first_line_end < 0: + return raw + opening = raw[:first_line_end].strip().lower() + if opening not in {"```", "```json"}: + return raw + body = raw[first_line_end + 1:].rstrip() + if body.endswith("```"): + body = body[:-3].rstrip() + return body + + +class TenantDreamingCompressor: + """Compress RAW memory with the tenant's configured default LLM.""" + + def __init__(self, tenant_id: str, user_id: str): + config = tenant_config_manager.get_model_config( + key=MODEL_CONFIG_MAPPING["llm"], tenant_id=tenant_id + ) + if not config: + raise RuntimeError("No tenant LLM is configured for Dreaming") + self.tenant_id = tenant_id + self.user_id = user_id + context_tokens = int( + config.get("max_input_tokens") + or config.get("context_window_tokens") + or 32_000 + ) + self.max_compression_input_chars = max(20_000, context_tokens * 3) + self.model = OpenAIModel( + model_id=get_model_name_from_config(config), + api_base=config.get("base_url", ""), + api_key=config.get("api_key", ""), + temperature=0.1, + top_p=0.9, + model_factory=config.get("model_factory"), + ssl_verify=config.get("ssl_verify", True), + display_name=config.get("display_name") or None, + timeout_seconds=config.get("timeout_seconds"), + stream=False, + ) + + def __call__( + self, request: DreamingCompressionRequest + ) -> DreamingCompressionOutput: + evidence_ids = sorted( + {evidence_id for unit in request.units for evidence_id in unit.evidence_ids} + ) + feedback = ", ".join(request.validation_feedback) or "none" + numeric_agent_id = ( + int(request.agent_id) + if request.agent_id and request.agent_id.isdigit() + else None + ) + metadata = AgentRunMetadata( + tenant_id=self.tenant_id, + user_id=self.user_id, + agent_id=numeric_agent_id, + extra_metadata={ + "dreaming_run_id": request.run_id, + "dreaming_agent_id": request.agent_id, + "dreaming_attempt": request.attempt, + }, + ) + with agent_monitoring_context(metadata): + input_limit = getattr(self, "max_compression_input_chars", 40_000) + + units = self._prepare_units_with_ids(request.units) + + if len(request.raw_content) > input_limit or len(units) > 12: + return self._map_reduce_extract( + request, units, evidence_ids, feedback + ) + + spans = self._extract_spans(request.raw_content, units, feedback) + + facts, span_feedback = self._validate_spans( + spans, request.raw_content, units + ) + if span_feedback: + raise ValueError( + f"Span validation failed: {span_feedback}" + ) + + facts.extend(self._required_literal_facts(units, len(facts))) + unique_facts = self._deterministic_dedup(facts) + self._require_source_coverage(unique_facts, units) + + output = self._format_facts( + unique_facts, request.max_chars, evidence_ids + ) + + # Stage 3: Coverage validation + all_fact_ids = [f["fact_id"] for f in unique_facts] + covered_fact_ids = output.metadata.get( + "covered_fact_ids", + self._count_covered_facts(output.content, unique_facts), + ) + output.metadata["covered_fact_ids"] = covered_fact_ids + + if len(all_fact_ids) > 0: + coverage = len(covered_fact_ids) / len(all_fact_ids) + if coverage < 0.95: + raise ValueError( + f"Fact coverage too low: {len(covered_fact_ids)}/{len(all_fact_ids)}={coverage:.2f}" + ) + + return output + + def _map_reduce_extract( + self, + request: DreamingCompressionRequest, + units: list[dict], + evidence_ids: list[str], + feedback: str, + ) -> DreamingCompressionOutput: + input_limit = getattr(self, "max_compression_input_chars", 40_000) + unit_models = request.units + chunks = self._chunk_units( + unit_models, + input_limit=max(10_000, input_limit // 2), + max_units=12, + ) + + all_facts = [] + fact_counter = 0 + for chunk in chunks: + chunk_units = [u for u in units if any( + m.unit_id == u["unit_id"] for m in chunk + )] + chunk_raw = "\n".join( + f"- {u['text'].strip()}" for u in chunk_units if u["text"].strip() + ) + spans = self._extract_spans(chunk_raw, chunk_units, feedback) + facts, span_fb = self._validate_spans(spans, chunk_raw, chunk_units) + if span_fb: + raise ValueError(f"Map chunk span validation failed: {span_fb}") + for f in facts: + f["fact_id"] = f"f{fact_counter:03d}" + fact_counter += 1 + all_facts.extend(facts) + + all_facts.extend( + self._required_literal_facts(units, fact_counter) + ) + unique_facts = self._deterministic_dedup(all_facts) + self._require_source_coverage(unique_facts, units) + output = self._format_facts(unique_facts, request.max_chars, evidence_ids) + + all_fact_ids = [f["fact_id"] for f in unique_facts] + covered_fact_ids = output.metadata.get( + "covered_fact_ids", + self._count_covered_facts(output.content, unique_facts), + ) + output.metadata["covered_fact_ids"] = covered_fact_ids + + if len(all_fact_ids) > 0: + coverage = len(covered_fact_ids) / len(all_fact_ids) + if coverage < 0.95: + raise ValueError( + f"Fact coverage too low: {len(covered_fact_ids)}/{len(all_fact_ids)}={coverage:.2f}" + ) + + return output + + @staticmethod + def _chunk_units( + units: list, *, input_limit: int, max_units: int = 12 + ) -> list[list]: + chunks: list[list] = [] + current: list = [] + current_chars = 0 + for unit in units: + unit_chars = len(unit.content) + 200 + if current and ( + current_chars + unit_chars > input_limit + or len(current) >= max_units + ): + chunks.append(current) + current = [] + current_chars = 0 + current.append(unit) + current_chars += unit_chars + if current: + chunks.append(current) + return chunks + + @staticmethod + def _prepare_units_with_ids( + units: list, + ) -> list[dict]: + """Pre-assign evidence IDs to each unit before LLM extraction.""" + prepared = [] + for unit in units: + prepared.append({ + "unit_id": unit.unit_id, + "text": unit.content, + "evidence_ids": list(unit.evidence_ids), + }) + return prepared + + @staticmethod + def _extract_spans_prompt(raw_content: str, units: list[dict]) -> str: + units_json = json.dumps( + [ + {"unit_id": unit["unit_id"], "text": unit["text"]} + for unit in units + ], + ensure_ascii=False, + ) + return ( + "Select all atomic facts from the memory-unit JSON below.\n\n" + "Rules:\n" + "1. Select every fact, including labels, identifiers, and numbers; " + "return one input unit_id and offsets indexing only that object's " + "text string (exclude JSON syntax and unit_id).\n" + "2. Do NOT generate or copy any text.\n" + "3. Each fact must be a contiguous substring of its selected unit.\n" + "4. Select the smallest complete atomic fact; split units that contain " + "multiple facts into separate spans.\n" + "5. Keep contradictory facts as separate spans.\n\n" + f"Memory units JSON:\n{units_json}\n\n" + "Return only a JSON array of objects with exactly these keys: " + "unit_id, start, end. start and end are character offsets in the " + "selected object's text value." + ) + + @staticmethod + def _compression_prompt(raw_content: str, units: list[dict]) -> str: + """Compatibility name for the approved information-extraction prompt.""" + return TenantDreamingCompressor._extract_spans_prompt(raw_content, units) + + def _extract_spans( + self, + raw_content: str, + units: list[dict], + feedback: str, + ) -> list[dict]: + """Call LLM to extract spans, parse JSON, return list of span dicts.""" + prompt = self._compression_prompt(raw_content, units) + if feedback and feedback != "none": + prompt += f"\n\nPrevious attempt feedback: {feedback}" + set_monitoring_operation("dreaming_semantic_compression_extract") + response = self.model.generate( + [ + { + "role": "system", + "content": ( + "You are an information extraction engine. " + "Your task is to select factual spans from RAW memory. " + "You do not summarize. You do not rewrite. You do not compress. " + "Return only unit-relative character offsets that exist exactly " + "in the selected source unit. " + "Output JSON only." + ), + }, + {"role": "user", "content": prompt}, + ] + ) + raw = _strip_json_fence(response.content) + return json.loads(raw) + + @staticmethod + def _validate_spans( + spans: list[dict], + raw_content: str, + units: list[dict], + ) -> tuple[list[dict], list[str]]: + """Validate spans and extract fact text. Returns (facts, feedback).""" + feedback = [] + valid_facts = [] + unit_evidence_map = {u["unit_id"]: u["evidence_ids"] for u in units} + unit_text_map = {u["unit_id"]: u["text"] for u in units} + + for i, span_obj in enumerate(spans): + start = span_obj.get("start") + end = span_obj.get("end") + + if start is None or end is None: + feedback.append(f"span_{i}_missing_offsets") + continue + + if not isinstance(start, int) or not isinstance(end, int): + feedback.append(f"span_{i}_non_integer_offsets") + continue + + unit_id = span_obj.get("unit_id") + legacy_unit_ids = span_obj.get("unit_ids") + if unit_id is None and isinstance(legacy_unit_ids, list): + if len(legacy_unit_ids) == 1: + unit_id = legacy_unit_ids[0] + if unit_id not in unit_text_map: + feedback.append(f"span_{i}_invalid_unit:{unit_id}") + continue + unit_text = unit_text_map[unit_id] + + if end > len(unit_text): + json_text_prefix = ( + json.dumps( + {"unit_id": unit_id, "text": ""}, + ensure_ascii=False, + )[:-2] + ) + normalized_start = start - len(json_text_prefix) + normalized_end = end - len(json_text_prefix) + if ( + normalized_start >= 0 + and normalized_start < normalized_end <= len(unit_text) + ): + start, end = normalized_start, normalized_end + + if start < 0 or end > len(unit_text): + feedback.append(f"span_{i}_out_of_bounds:{start},{end}") + continue + + if start >= end: + feedback.append(f"span_{i}_invalid_range:{start}>={end}") + continue + + fact_text = unit_text[start:end].strip() + + if not fact_text: + feedback.append(f"span_{i}_empty_text:{start},{end}") + continue + + unit_ids = [unit_id] + evidence_ids = sorted(set(unit_evidence_map.get(unit_id, []))) + + valid_facts.append({ + "fact_id": f"f{i:03d}", + "unit_ids": unit_ids, + "span": {"start": start, "end": end}, + "text": fact_text, + "evidence_ids": evidence_ids, + }) + + return valid_facts, feedback + + @staticmethod + def _deterministic_dedup(facts: list[dict]) -> list[dict]: + """Remove exact duplicates while preserving their source attribution.""" + by_text: dict[str, dict] = {} + order: list[str] = [] + + for fact_obj in facts: + normalized = re.sub(r"\s+", " ", fact_obj["text"].strip()) + if normalized in by_text: + retained = by_text[normalized] + retained["unit_ids"] = sorted( + set(retained["unit_ids"]) | set(fact_obj["unit_ids"]) + ) + retained["evidence_ids"] = sorted( + set(retained["evidence_ids"]) | set(fact_obj["evidence_ids"]) + ) + continue + by_text[normalized] = { + **fact_obj, + "unit_ids": sorted(set(fact_obj["unit_ids"])), + "evidence_ids": sorted(set(fact_obj["evidence_ids"])), + } + order.append(normalized) + + return [by_text[text] for text in order] + + @staticmethod + def _required_literal_facts( + units: list[dict], start_index: int + ) -> list[dict]: + """Extract validation-critical literals from authoritative unit text.""" + patterns = ( + r"https?://[^\s)\]}>,]+", + r"[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}", + r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-" + r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b", + r"(? None: + """Reject extraction that represents fewer than 95% of source units.""" + source_unit_ids = {unit["unit_id"] for unit in units} + covered_unit_ids = { + unit_id for fact in facts for unit_id in fact["unit_ids"] + } + if not source_unit_ids: + return + coverage = len(covered_unit_ids) / len(source_unit_ids) + if coverage < 0.95: + raise ValueError( + "Source unit coverage too low: " + f"{len(covered_unit_ids)}/{len(source_unit_ids)}={coverage:.2f}" + ) + + def _format_facts( + self, + facts: list[dict], + max_chars: int, + evidence_ids: list[str], + ) -> DreamingCompressionOutput: + """Format facts as bullet list. LLM only if over limit.""" + content = "\n".join(f"- {f['text']}" for f in facts) + + if len(content) <= max_chars: + return DreamingCompressionOutput( + content=content, + evidence_ids=evidence_ids, + metadata={ + "all_fact_ids": [f["fact_id"] for f in facts], + "covered_fact_ids": [f["fact_id"] for f in facts], + "fact_to_units_map": {f["fact_id"]: f["unit_ids"] for f in facts}, + }, + ) + + return self._lossless_formatting(facts, max_chars, evidence_ids) + + def _lossless_formatting( + self, + facts: list[dict], + max_chars: int, + evidence_ids: list[str], + ) -> DreamingCompressionOutput: + """LLM-based character-level shortening. No semantic changes.""" + facts_text = "\n".join( + f"[{fact['fact_id']}] {fact['text']}" for fact in facts + ) + prompt = ( + "You are a lossless formatter.\n\n" + "Input contains validated facts.\n" + "Your task: reduce character count while preserving every fact.\n\n" + "Allowed:\n" + "- Remove redundant whitespace\n" + "- Shorten connective words (e.g., 'in order to' → 'to')\n" + "- Change bullet formatting\n\n" + "Forbidden:\n" + "- Merge facts\n" + "- Reorder facts if it changes meaning\n" + "- Generalize facts\n" + "- Remove examples\n" + "- Remove identifiers\n" + "- Combine facts\n\n" + f"The following facts total {len(facts_text)} characters.\n" + f"Format them to fit within {max_chars} characters.\n\n" + f"Facts:\n{facts_text}\n\n" + "Return strict JSON with one entry for every supplied fact_id:\n" + '{"facts": [{"fact_id": "f001", "text": "shortened fact"}]}\n' + "Each fact_id must appear exactly once. Never combine multiple fact_ids." + ) + set_monitoring_operation("dreaming_semantic_compression_format") + response = self.model.generate( + [ + { + "role": "system", + "content": ( + "You are a lossless formatter. " + "You shorten text without changing meaning. " + "Output JSON only." + ), + }, + {"role": "user", "content": prompt}, + ] + ) + raw = _strip_json_fence(response.content) + payload = json.loads(raw) + formatted_facts = payload.get("facts") + if not isinstance(formatted_facts, list): + raise ValueError("Lossless formatting response must contain a facts list") + expected_ids = [fact["fact_id"] for fact in facts] + returned_ids = [ + item.get("fact_id") for item in formatted_facts + if isinstance(item, dict) + ] + if len(returned_ids) != len(set(returned_ids)): + raise ValueError("Lossless formatting returned duplicate fact_ids") + if set(returned_ids) != set(expected_ids): + raise ValueError( + "Lossless formatting fact_ids do not match the extracted facts" + ) + text_by_id = {} + for item in formatted_facts: + text = item.get("text") + if not isinstance(text, str) or not text.strip(): + raise ValueError("Lossless formatting returned an empty fact") + text_by_id[item["fact_id"]] = text.strip() + content = "\n".join(f"- {text_by_id[fact_id]}" for fact_id in expected_ids) + return DreamingCompressionOutput( + content=content, + evidence_ids=evidence_ids, + metadata={ + "all_fact_ids": expected_ids, + "covered_fact_ids": returned_ids, + "fact_to_units_map": {f["fact_id"]: f["unit_ids"] for f in facts}, + }, + ) + + @staticmethod + def _count_covered_facts( + output_content: str, + facts: list[dict], + ) -> list[str]: + """Return fact_ids whose text appears in output_content.""" + covered = [] + for fact_obj in facts: + if fact_obj["text"] in output_content: + covered.append(fact_obj["fact_id"]) + return covered diff --git a/backend/services/memory_dreaming_scheduler.py b/backend/services/memory_dreaming_scheduler.py index 4f5634523..3382e4d21 100644 --- a/backend/services/memory_dreaming_scheduler.py +++ b/backend/services/memory_dreaming_scheduler.py @@ -1,411 +1,138 @@ -"""Dreaming consolidation runner for the Memory Architecture (Phase 2). - -Dreaming promotes agent short-term memories into user long-term memory. It -runs in three phases: - -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 +"""Backend adapter for the SDK's durable lease scheduler — dreaming jobs.""" +import asyncio 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 typing import Any, Dict, Hashable 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, + DREAMING_SCHEDULER_ENABLED, + DREAMING_SCHEDULER_LEASE_SECONDS, + DREAMING_SCHEDULER_MAX_CONCURRENCY, + DREAMING_SCHEDULER_POLL_SECONDS, ) -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, -} +from database import memory_dreaming_db +from nexent.scheduler import ClaimedJob, ExecutionLease, LeaseScheduler, SchedulerConfig -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()} +logger = logging.getLogger("memory_dreaming.scheduler") -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)) +class DreamingLeaseStore: + """Adapt synchronous PostgreSQL operations to the async scheduler contract.""" + @staticmethod + def _materialize_and_claim( + owner_id: str, limit: int, lease_seconds: float + ) -> Dict[str, Any] | None: + memory_dreaming_db.materialize_due_schedules(limit) + return memory_dreaming_db.claim_queued(owner_id, lease_seconds) -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 [] + async def recover(self) -> None: + await asyncio.to_thread(memory_dreaming_db.recover_stale) - 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, - }, + async def claim_due( + self, + owner_id: str, + limit: int, + lease_seconds: float, + ) -> list[ClaimedJob[Dict[str, Any]]]: + row = await asyncio.to_thread( + self._materialize_and_claim, + owner_id, + limit, + lease_seconds, ) - memory_record_db.apply_dreaming_phase( - memory_id, tenant_id, phase="light" + if row is None: + return [] + return [ClaimedJob(job_id=row["run_id"], payload=row)] + + async def renew(self, job_id: Hashable, owner_id: str, lease_seconds: float) -> bool: + return await asyncio.to_thread( + memory_dreaming_db.renew_lease, + int(job_id), + owner_id, + lease_seconds, ) - 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", - "的", "了", "是", "在", "和", "与", "及", "或", "我", "你", "他", "她", "它", - "我们", "你们", "他们", "这", "那", "这个", "那个", -} + async def release(self, job_id: Hashable, owner_id: str) -> bool: + return await asyncio.to_thread( + memory_dreaming_db.release_lease, + int(job_id), + owner_id, + ) -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 - ] +async def execute_dreaming( + job: ClaimedJob[Dict[str, Any]], + lease: ExecutionLease, +) -> None: + """Executor callback invoked by the SDK scheduler for each claimed dreaming job.""" + # Lazy import to avoid circular dependencies at module load time. + from services.memory_dreaming_service import get_memory_dreaming_service -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. + payload = job.payload + tenant_id = payload["tenant_id"] + user_id = payload["user_id"] + agent_id = payload["agent_id"] + trigger_source = payload.get("trigger_source", "scheduler") - 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"], + try: + await asyncio.to_thread( + get_memory_dreaming_service().run, + tenant_id=tenant_id, + user_id=user_id, + agent_id=agent_id, + run_id=int(job.job_id), + trigger_source=trigger_source, + ) + logger.info( + "Dreaming job completed: run_id=%s tenant=%s user=%s agent=%s", + job.job_id, tenant_id, - {"concept_tags": merged}, + user_id, + agent_id, ) - memory_record_db.apply_dreaming_phase( - row["memory_id"], tenant_id, phase="rem" + except Exception: + logger.exception( + "Dreaming job failed: run_id=%s tenant=%s user=%s agent=%s", + job.job_id, + tenant_id, + user_id, + agent_id, ) - touched += 1 - return touched + raise -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. +class DreamingScheduler: + """Application lifecycle wrapper around the reusable SDK scheduler.""" - 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, - ) - 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", - } + def __init__(self) -> None: + self._scheduler = LeaseScheduler( + store=DreamingLeaseStore(), + executor=execute_dreaming, + config=SchedulerConfig( + poll_interval_seconds=DREAMING_SCHEDULER_POLL_SECONDS, + lease_seconds=DREAMING_SCHEDULER_LEASE_SECONDS, + max_concurrency=DREAMING_SCHEDULER_MAX_CONCURRENCY, + ), ) - 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. + @property + def instance_id(self) -> str: + return self._scheduler.owner_id - Args: - timeout_seconds: Soft cap on wall-clock runtime per call. Iteration - stops once the deadline is reached; partial state is returned. + @property + def is_running(self) -> bool: + return self._scheduler.is_running - 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) + async def start(self) -> None: + if not DREAMING_SCHEDULER_ENABLED: + logger.info("Dreaming scheduler disabled") + return + await self._scheduler.start() - # ``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": [], - } + async def stop(self) -> None: + await self._scheduler.stop() - 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 +dreaming_scheduler = DreamingScheduler() diff --git a/backend/services/memory_dreaming_service.py b/backend/services/memory_dreaming_service.py new file mode 100644 index 000000000..a77346ec0 --- /dev/null +++ b/backend/services/memory_dreaming_service.py @@ -0,0 +1,396 @@ +"""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 ( + DREAMING_COMPRESSION_MAX_ATTEMPTS, + DREAMING_LONG_TERM_MAX_CHARS, + DREAMING_SOURCE_LIMIT, + 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 ( + DreamingMemoryUnit, + DreamingThresholds, + build_dreaming_version, + build_candidate, + select_candidates, + units_from_decisions, +) +from services.memory_record_service import get_memory_record_service + +logger = logging.getLogger("memory_dreaming_service") +USER_DREAMING_SCOPE = "__user__" + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +class DreamingRunError(RuntimeError): + pass + + +class DreamingConflictError(RuntimeError): + pass + + +class MemoryDreamingService: + def __init__(self, record_service: Any = None, compressor: Any = None): + self.record_service = record_service or get_memory_record_service() + self.compressor = compressor + + 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, + None if agent_id == USER_DREAMING_SCOPE else 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=None if agent_id == USER_DREAMING_SCOPE else agent_id, + layer="agent", + memory_type="short_term", + status="active", + limit=None, + ) + 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 _build_version( + self, + tenant_id: str, + user_id: str, + agent_id: str, + run_id: int, + decisions: List[Any], + config_snapshot: Optional[Dict[str, Any]] = None, + ) -> Optional[Dict[str, Any]]: + active = memory_dreaming_db.get_active_version(tenant_id, user_id, agent_id) + parent_units = [ + DreamingMemoryUnit.model_validate(unit) + for unit in (active or {}).get("published_units", []) + ] + parent_unit_ids = {unit.unit_id for unit in parent_units} + parent_evidence_ids = { + evidence_id for unit in parent_units for evidence_id in unit.evidence_ids + } + new_units = units_from_decisions( + decisions, + source_limit=DREAMING_SOURCE_LIMIT, + excluded_evidence_ids=parent_evidence_ids, + ) + new_units = [ + unit + for unit in new_units + if unit.unit_id not in parent_unit_ids + and not set(unit.evidence_ids).issubset(parent_evidence_ids) + ] + if not new_units: + return None + result = build_dreaming_version( + parent_units=parent_units, + new_units=new_units, + max_chars=DREAMING_LONG_TERM_MAX_CHARS, + compressor=self.compressor or self._tenant_compressor(tenant_id, user_id), + max_attempts=DREAMING_COMPRESSION_MAX_ATTEMPTS, + run_id=run_id, + agent_id=agent_id, + ) + return memory_dreaming_db.create_and_activate_version( + tenant_id=tenant_id, + user_id=user_id, + agent_id=agent_id, + run_id=run_id, + parent_version_id=(active or {}).get("version_id"), + raw_content=result.raw_content, + published_content=result.published_content, + published_units=[ + unit.model_dump(mode="json") for unit in result.published_units + ], + source_evidence_ids=sorted( + { + evidence_id + for unit in [*parent_units, *new_units] + for evidence_id in unit.evidence_ids + } + ), + config_snapshot=config_snapshot + or { + "source_limit": DREAMING_SOURCE_LIMIT, + "long_term_max_chars": DREAMING_LONG_TERM_MAX_CHARS, + "compression_max_attempts": DREAMING_COMPRESSION_MAX_ATTEMPTS, + }, + raw_char_count=result.raw_char_count, + published_char_count=result.published_char_count, + compression_status=result.compression_status, + compression_attempts=result.compression_attempts, + omitted_evidence_ids=result.omitted_evidence_ids, + mechanical_truncation=result.mechanical_truncation, + compression_audit=result.compression_audit, + ) + + @staticmethod + def _tenant_compressor(tenant_id: str, user_id: str): + instance = None + + def compress(request): + nonlocal instance + if instance is None: + from services.memory_dreaming_compressor import TenantDreamingCompressor + + instance = TenantDreamingCompressor(tenant_id, user_id) + return instance(request) + + return compress + + 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, + run_id: Optional[int] = None, + trigger_source: str = "manual", + ) -> 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") + if run_id is None: + if trigger_source == "manual": + run_id = memory_dreaming_db.create_audit(tenant_id, user_id, agent_id) + else: + run_id = memory_dreaming_db.create_audit( + tenant_id, + user_id, + agent_id, + trigger_source=trigger_source, + ) + else: + memory_dreaming_db.update_audit( + run_id, {"status": "running", "current_phase": "light"} + ) + 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, + ) + memory_dreaming_db.update_audit( + run_id, {"current_phase": "compression"} + ) + version = self._build_version( + tenant_id, + user_id, + agent_id, + run_id, + decisions, + config_snapshot={ + "window_days": window_days, + "min_score": min_score, + "min_recall_count": min_recall_count, + "min_unique_queries": min_unique_queries, + "source_limit": DREAMING_SOURCE_LIMIT, + "long_term_max_chars": DREAMING_LONG_TERM_MAX_CHARS, + "compression_max_attempts": (DREAMING_COMPRESSION_MAX_ATTEMPTS), + }, + ) + results = [ + { + "memory_id": decision.candidate.memory_id, + "score": decision.score, + "evidence_ids": [str(decision.candidate.memory_id)], + "event": "SELECT" if decision.promote else "DEFER", + "reason": decision.reason, + "archive_suggested": decision.archive_suggested, + } + for decision in decisions + ] + promoted_count = sum(decision.promote for decision in decisions) + 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, + "version": version, + } + 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, + ) + + def list_versions( + self, tenant_id: str, user_id: str, *, agent_id: str, limit: int = 100 + ) -> List[Dict[str, Any]]: + return memory_dreaming_db.list_versions( + tenant_id, user_id, agent_id=agent_id, limit=limit + ) + + def activate_version( + self, + tenant_id: str, + user_id: str, + *, + agent_id: str, + version_id: int, + actor_user_id: Optional[str] = None, + expected_active_version_id: Optional[int] = None, + ) -> Optional[Dict[str, Any]]: + with memory_dreaming_db.try_scope_lock( + tenant_id, user_id, agent_id + ) as acquired: + if not acquired: + raise DreamingConflictError("Dreaming scope is busy") + active = memory_dreaming_db.get_active_version(tenant_id, user_id, agent_id) + if ( + expected_active_version_id is not None + and (active or {}).get("version_id") != expected_active_version_id + ): + raise DreamingConflictError( + "Active Dreaming version changed; refresh and retry" + ) + return memory_dreaming_db.activate_version( + tenant_id, + user_id, + agent_id, + version_id, + actor_user_id=actor_user_id, + ) + + +_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/context_utils.py b/backend/utils/context_utils.py index 9e4d12097..a7c669457 100644 --- a/backend/utils/context_utils.py +++ b/backend/utils/context_utils.py @@ -391,7 +391,19 @@ def add_system( inputs.append(ContextItemInput( id=f"memory:{index}", type=ContextItemType.MEMORY, content=payload, source=(f"memory:{memory_search_query or 'run'}",), priority=90, - metadata={"render_group": "memory", "language": language, "authority": "retrieved"}, + metadata={ + "render_group": "memory", + "language": language, + "authority": "retrieved", + **( + { + "version_id": payload["dreaming_version_id"], + "memory_type": "long_term", + } + if payload.get("dreaming_version_id") is not None + else {} + ), + }, )) if duty: diff --git a/backend/utils/memory_utils.py b/backend/utils/memory_utils.py new file mode 100644 index 000000000..3bac96077 --- /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/k8s/deploy.sh b/deploy/k8s/deploy.sh index b765845b2..564baf41a 100755 --- a/deploy/k8s/deploy.sh +++ b/deploy/k8s/deploy.sh @@ -492,7 +492,7 @@ render_k8s_runtime_config_values() { printf ' celeryWorkerPrefetchMultiplier: %s\n' "$(yaml_quote "$(env_or_default CELERY_WORKER_PREFETCH_MULTIPLIER "1")")" printf ' celeryTaskTimeLimit: %s\n' "$(yaml_quote "$(env_or_default CELERY_TASK_TIME_LIMIT "3600")")" printf ' elasticsearchRequestTimeout: %s\n' "$(yaml_quote "$(env_or_default ELASTICSEARCH_REQUEST_TIMEOUT "30")")" - printf ' queues: %s\n' "$(yaml_quote "$(env_or_default QUEUES "process_q,forward_q")")" + printf ' queues: %s\n' "$(yaml_quote "$(env_or_default QUEUES "process_q,process_part_q,forward_q,dreaming_q")")" printf ' workerName: %s\n' "$(yaml_quote "$(env_or_default WORKER_NAME "")")" printf ' workerConcurrency: %s\n' "$(yaml_quote "$(env_or_default WORKER_CONCURRENCY "4")")" echo " oauth:" diff --git a/deploy/k8s/helm/nexent/charts/nexent-common/values.yaml b/deploy/k8s/helm/nexent/charts/nexent-common/values.yaml index 446bf9cd3..a43784bd2 100644 --- a/deploy/k8s/helm/nexent/charts/nexent-common/values.yaml +++ b/deploy/k8s/helm/nexent/charts/nexent-common/values.yaml @@ -124,7 +124,7 @@ config: celeryWorkerPrefetchMultiplier: "1" celeryTaskTimeLimit: "3600" elasticsearchRequestTimeout: "30" - queues: "process_q,forward_q" + queues: "process_q,process_part_q,forward_q,dreaming_q" workerName: "" workerConcurrency: "4" telemetry: diff --git a/deploy/sql/init.sql b/deploy/sql/init.sql index 9d64ab242..0de37ce94 100644 --- a/deploy/sql/init.sql +++ b/deploy/sql/init.sql @@ -230,6 +230,58 @@ COMMENT ON COLUMN "knowledge_record_t"."updated_by" IS 'Last updater ID, audit f COMMENT ON COLUMN "knowledge_record_t"."created_by" IS 'Creator ID, audit field'; COMMENT ON TABLE "knowledge_record_t" IS 'Records knowledge base description and status information'; +-- Create the ag_prompt_template_t table +CREATE TABLE IF NOT EXISTS nexent.ag_prompt_template_t ( + template_id SERIAL PRIMARY KEY, + template_name VARCHAR(100) NOT NULL, + description VARCHAR(500), + template_type VARCHAR(50) NOT NULL DEFAULT 'agent_generate', + tenant_id VARCHAR(100) NOT NULL, + user_id VARCHAR(100) NOT NULL, + template_content_zh JSONB NOT NULL, + template_content_en JSONB, + create_time TIMESTAMP WITHOUT TIME ZONE DEFAULT CURRENT_TIMESTAMP, + update_time TIMESTAMP WITHOUT TIME ZONE DEFAULT CURRENT_TIMESTAMP, + created_by VARCHAR(100), + updated_by VARCHAR(100), + delete_flag VARCHAR(1) DEFAULT 'N' +); + +CREATE OR REPLACE FUNCTION update_ag_prompt_template_update_time() +RETURNS TRIGGER AS $$ +BEGIN + NEW.update_time = CURRENT_TIMESTAMP; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER update_ag_prompt_template_update_time_trigger +BEFORE UPDATE ON nexent.ag_prompt_template_t +FOR EACH ROW +EXECUTE FUNCTION update_ag_prompt_template_update_time(); + +COMMENT ON TABLE nexent.ag_prompt_template_t IS 'Prompt template table for user-defined business logic generation prompts'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.template_id IS 'Prompt template ID'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.template_name IS 'Prompt template name'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.description IS 'Prompt template description'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.template_type IS 'Prompt template type'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.tenant_id IS 'Tenant ID'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.user_id IS 'User ID'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.template_content_zh IS 'Chinese prompt template content'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.template_content_en IS 'English prompt template content'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.create_time IS 'Creation time'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.update_time IS 'Update time'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.created_by IS 'Creator'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.updated_by IS 'Updater'; +COMMENT ON COLUMN nexent.ag_prompt_template_t.delete_flag IS 'Whether it is deleted. Optional values: Y/N'; + +CREATE UNIQUE INDEX IF NOT EXISTS uq_prompt_template_user_name_active +ON nexent.ag_prompt_template_t (tenant_id, user_id, template_name) +WHERE delete_flag = 'N'; + +CREATE INDEX IF NOT EXISTS idx_ag_prompt_template_t_user +ON nexent.ag_prompt_template_t (tenant_id, user_id, template_type); + -- Create the ag_tool_info_t table CREATE TABLE IF NOT EXISTS nexent.ag_tool_info_t ( tool_id SERIAL PRIMARY KEY NOT NULL, @@ -718,3 +770,180 @@ 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, + lock_owner VARCHAR(100), + lock_until TIMESTAMP, + 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); + +CREATE TABLE IF NOT EXISTS nexent.memory_dreaming_schedule_t ( + schedule_id BIGSERIAL PRIMARY KEY, + tenant_id VARCHAR(100) NOT NULL, + user_id VARCHAR(100) NOT NULL, + agent_id VARCHAR(100) NOT NULL, + enabled BOOLEAN NOT NULL DEFAULT FALSE, + rule_type VARCHAR(20) NOT NULL DEFAULT 'CRON', + timezone VARCHAR(100) NOT NULL DEFAULT 'Asia/Shanghai', + start_at TIMESTAMP NOT NULL, + cron_expr VARCHAR(100), + interval_seconds INTEGER, + next_fire_at TIMESTAMP, + last_fire_at TIMESTAMP, + fire_count INTEGER NOT NULL DEFAULT 0, + 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', + CONSTRAINT ck_memory_dreaming_schedule_rule CHECK ( + (rule_type = 'CRON' AND cron_expr IS NOT NULL AND interval_seconds IS NULL) + OR (rule_type = 'INTERVAL' AND cron_expr IS NULL AND interval_seconds >= 3600) + ) +); +CREATE UNIQUE INDEX IF NOT EXISTS uq_memory_dreaming_schedule_scope + ON nexent.memory_dreaming_schedule_t (tenant_id, user_id, agent_id); +CREATE INDEX IF NOT EXISTS idx_memory_dreaming_schedule_due + ON nexent.memory_dreaming_schedule_t (enabled, next_fire_at) + WHERE delete_flag = 'N'; + +-- Immutable, switchable Dreaming long-term memory versions. +CREATE TABLE IF NOT EXISTS nexent.memory_dreaming_version_t ( + version_id BIGSERIAL PRIMARY KEY, + tenant_id VARCHAR(100) NOT NULL, + user_id VARCHAR(100) NOT NULL, + agent_id VARCHAR(100) NOT NULL, + version_no INTEGER NOT NULL, + parent_version_id BIGINT, + run_id BIGINT NOT NULL, + is_active BOOLEAN NOT NULL DEFAULT FALSE, + raw_content TEXT NOT NULL, + published_content TEXT NOT NULL, + published_units JSONB NOT NULL DEFAULT '[]'::jsonb, + source_evidence_ids JSONB NOT NULL DEFAULT '[]'::jsonb, + config_snapshot JSONB NOT NULL DEFAULT '{}'::jsonb, + raw_char_count INTEGER NOT NULL, + published_char_count INTEGER NOT NULL, + compression_status VARCHAR(30) NOT NULL, + compression_attempts INTEGER NOT NULL DEFAULT 0, + compression_audit JSONB NOT NULL DEFAULT '[]'::jsonb, + omitted_evidence_ids JSONB NOT NULL DEFAULT '[]'::jsonb, + mechanical_truncation BOOLEAN NOT NULL DEFAULT FALSE, + 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 UNIQUE INDEX IF NOT EXISTS idx_memory_dreaming_version_scope + ON nexent.memory_dreaming_version_t + (tenant_id, user_id, agent_id, version_no); +CREATE UNIQUE INDEX IF NOT EXISTS uq_memory_dreaming_version_active_scope + ON nexent.memory_dreaming_version_t + (tenant_id, user_id, agent_id) + WHERE is_active AND delete_flag = 'N'; +CREATE UNIQUE INDEX IF NOT EXISTS uq_memory_dreaming_version_run + ON nexent.memory_dreaming_version_t (run_id); +CREATE INDEX IF NOT EXISTS idx_memory_dreaming_version_history + ON nexent.memory_dreaming_version_t + (tenant_id, user_id, agent_id, create_time DESC); +CREATE OR REPLACE FUNCTION nexent.prevent_memory_dreaming_version_content_update() +RETURNS TRIGGER AS $$ +BEGIN + IF OLD.version_no IS DISTINCT FROM NEW.version_no + OR OLD.parent_version_id IS DISTINCT FROM NEW.parent_version_id + OR OLD.run_id IS DISTINCT FROM NEW.run_id + OR OLD.raw_content IS DISTINCT FROM NEW.raw_content + OR OLD.published_content IS DISTINCT FROM NEW.published_content + OR OLD.published_units IS DISTINCT FROM NEW.published_units + OR OLD.source_evidence_ids IS DISTINCT FROM NEW.source_evidence_ids + OR OLD.config_snapshot IS DISTINCT FROM NEW.config_snapshot + OR OLD.raw_char_count IS DISTINCT FROM NEW.raw_char_count + OR OLD.published_char_count IS DISTINCT FROM NEW.published_char_count + OR OLD.compression_status IS DISTINCT FROM NEW.compression_status + OR OLD.compression_attempts IS DISTINCT FROM NEW.compression_attempts + OR OLD.compression_audit IS DISTINCT FROM NEW.compression_audit + OR OLD.omitted_evidence_ids IS DISTINCT FROM NEW.omitted_evidence_ids + OR OLD.mechanical_truncation IS DISTINCT FROM NEW.mechanical_truncation THEN + RAISE EXCEPTION 'Dreaming version content is immutable'; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS trg_memory_dreaming_version_immutable + ON nexent.memory_dreaming_version_t; +CREATE TRIGGER trg_memory_dreaming_version_immutable +BEFORE UPDATE ON nexent.memory_dreaming_version_t +FOR EACH ROW EXECUTE FUNCTION nexent.prevent_memory_dreaming_version_content_update(); + +CREATE TABLE IF NOT EXISTS nexent.memory_dreaming_activation_audit_t ( + activation_id BIGSERIAL PRIMARY KEY, + tenant_id VARCHAR(100) NOT NULL, + user_id VARCHAR(100) NOT NULL, + agent_id VARCHAR(100) NOT NULL, + actor_user_id VARCHAR(100) NOT NULL, + from_version_id BIGINT, + to_version_id BIGINT NOT NULL, + reason VARCHAR(100) NOT NULL DEFAULT 'user_switch', + 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_activation_scope + ON nexent.memory_dreaming_activation_audit_t + (tenant_id, user_id, agent_id, create_time DESC); + +CREATE TABLE IF NOT EXISTS nexent.role_permission_t ( + role_permission_id SERIAL PRIMARY KEY, + user_role VARCHAR(30) NOT NULL, + permission_category VARCHAR(30), + permission_type VARCHAR(30), + permission_subtype VARCHAR(30), + parent_key VARCHAR(100) +); + +INSERT INTO nexent.role_permission_t ( + role_permission_id, + user_role, + permission_category, + permission_type, + permission_subtype +) +VALUES + (1004, 'SU', 'RESOURCE', 'DREAMING', 'VIEW_TENANT'), + (1005, 'SU', 'RESOURCE', 'DREAMING', 'EDIT_TENANT'), + (1116, 'ADMIN', 'RESOURCE', 'DREAMING', 'VIEW_TENANT'), + (1117, 'ADMIN', 'RESOURCE', 'DREAMING', 'EDIT_TENANT'), + (1514, 'ASSET_OWNER', 'RESOURCE', 'DREAMING', 'VIEW_TENANT'), + (1515, 'ASSET_OWNER', 'RESOURCE', 'DREAMING', 'EDIT_TENANT') +ON CONFLICT (role_permission_id) DO UPDATE SET + user_role = EXCLUDED.user_role, + permission_category = EXCLUDED.permission_category, + permission_type = EXCLUDED.permission_type, + permission_subtype = EXCLUDED.permission_subtype; 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 000000000..4986a2b95 --- /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/deploy/sql/migrations/v2.4.0_0723_add_memory_dreaming_version.sql b/deploy/sql/migrations/v2.4.0_0723_add_memory_dreaming_version.sql new file mode 100644 index 000000000..631315ebc --- /dev/null +++ b/deploy/sql/migrations/v2.4.0_0723_add_memory_dreaming_version.sql @@ -0,0 +1,124 @@ +-- Immutable Dreaming long-term memory versions and active-version pointer. +SET search_path TO nexent; +BEGIN; + +CREATE TABLE IF NOT EXISTS nexent.memory_dreaming_version_t ( + version_id BIGSERIAL PRIMARY KEY, + tenant_id VARCHAR(100) NOT NULL, + user_id VARCHAR(100) NOT NULL, + agent_id VARCHAR(100) NOT NULL, + version_no INTEGER NOT NULL, + parent_version_id BIGINT, + run_id BIGINT NOT NULL, + is_active BOOLEAN NOT NULL DEFAULT FALSE, + raw_content TEXT NOT NULL, + published_content TEXT NOT NULL, + published_units JSONB NOT NULL DEFAULT '[]'::jsonb, + source_evidence_ids JSONB NOT NULL DEFAULT '[]'::jsonb, + config_snapshot JSONB NOT NULL DEFAULT '{}'::jsonb, + raw_char_count INTEGER NOT NULL, + published_char_count INTEGER NOT NULL, + compression_status VARCHAR(30) NOT NULL, + compression_attempts INTEGER NOT NULL DEFAULT 0, + compression_audit JSONB NOT NULL DEFAULT '[]'::jsonb, + omitted_evidence_ids JSONB NOT NULL DEFAULT '[]'::jsonb, + mechanical_truncation BOOLEAN NOT NULL DEFAULT FALSE, + 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' +); + +ALTER TABLE nexent.memory_dreaming_version_t + ADD COLUMN IF NOT EXISTS compression_audit JSONB NOT NULL DEFAULT '[]'::jsonb; +ALTER TABLE nexent.memory_dreaming_version_t + ADD COLUMN IF NOT EXISTS source_evidence_ids JSONB NOT NULL DEFAULT '[]'::jsonb, + ADD COLUMN IF NOT EXISTS config_snapshot JSONB NOT NULL DEFAULT '{}'::jsonb; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_memory_dreaming_version_scope + ON nexent.memory_dreaming_version_t + (tenant_id, user_id, agent_id, version_no); + +CREATE UNIQUE INDEX IF NOT EXISTS uq_memory_dreaming_version_active_scope + ON nexent.memory_dreaming_version_t + (tenant_id, user_id, agent_id) + WHERE is_active AND delete_flag = 'N'; + +CREATE UNIQUE INDEX IF NOT EXISTS uq_memory_dreaming_version_run + ON nexent.memory_dreaming_version_t (run_id); + +CREATE INDEX IF NOT EXISTS idx_memory_dreaming_version_history + ON nexent.memory_dreaming_version_t + (tenant_id, user_id, agent_id, create_time DESC); + +CREATE OR REPLACE FUNCTION nexent.prevent_memory_dreaming_version_content_update() +RETURNS TRIGGER AS $$ +BEGIN + IF OLD.version_no IS DISTINCT FROM NEW.version_no + OR OLD.parent_version_id IS DISTINCT FROM NEW.parent_version_id + OR OLD.run_id IS DISTINCT FROM NEW.run_id + OR OLD.raw_content IS DISTINCT FROM NEW.raw_content + OR OLD.published_content IS DISTINCT FROM NEW.published_content + OR OLD.published_units IS DISTINCT FROM NEW.published_units + OR OLD.source_evidence_ids IS DISTINCT FROM NEW.source_evidence_ids + OR OLD.config_snapshot IS DISTINCT FROM NEW.config_snapshot + OR OLD.raw_char_count IS DISTINCT FROM NEW.raw_char_count + OR OLD.published_char_count IS DISTINCT FROM NEW.published_char_count + OR OLD.compression_status IS DISTINCT FROM NEW.compression_status + OR OLD.compression_attempts IS DISTINCT FROM NEW.compression_attempts + OR OLD.compression_audit IS DISTINCT FROM NEW.compression_audit + OR OLD.omitted_evidence_ids IS DISTINCT FROM NEW.omitted_evidence_ids + OR OLD.mechanical_truncation IS DISTINCT FROM NEW.mechanical_truncation THEN + RAISE EXCEPTION 'Dreaming version content is immutable'; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS trg_memory_dreaming_version_immutable + ON nexent.memory_dreaming_version_t; +CREATE TRIGGER trg_memory_dreaming_version_immutable +BEFORE UPDATE ON nexent.memory_dreaming_version_t +FOR EACH ROW EXECUTE FUNCTION nexent.prevent_memory_dreaming_version_content_update(); + +CREATE TABLE IF NOT EXISTS nexent.memory_dreaming_activation_audit_t ( + activation_id BIGSERIAL PRIMARY KEY, + tenant_id VARCHAR(100) NOT NULL, + user_id VARCHAR(100) NOT NULL, + agent_id VARCHAR(100) NOT NULL, + actor_user_id VARCHAR(100) NOT NULL, + from_version_id BIGINT, + to_version_id BIGINT NOT NULL, + reason VARCHAR(100) NOT NULL DEFAULT 'user_switch', + 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_activation_scope + ON nexent.memory_dreaming_activation_audit_t + (tenant_id, user_id, agent_id, create_time DESC); + +INSERT INTO nexent.role_permission_t ( + role_permission_id, + user_role, + permission_category, + permission_type, + permission_subtype +) +VALUES + (1004, 'SU', 'RESOURCE', 'DREAMING', 'VIEW_TENANT'), + (1005, 'SU', 'RESOURCE', 'DREAMING', 'EDIT_TENANT'), + (1116, 'ADMIN', 'RESOURCE', 'DREAMING', 'VIEW_TENANT'), + (1117, 'ADMIN', 'RESOURCE', 'DREAMING', 'EDIT_TENANT'), + (1514, 'ASSET_OWNER', 'RESOURCE', 'DREAMING', 'VIEW_TENANT'), + (1515, 'ASSET_OWNER', 'RESOURCE', 'DREAMING', 'EDIT_TENANT') +ON CONFLICT (role_permission_id) DO UPDATE SET + user_role = EXCLUDED.user_role, + permission_category = EXCLUDED.permission_category, + permission_type = EXCLUDED.permission_type, + permission_subtype = EXCLUDED.permission_subtype; + +COMMIT; diff --git a/deploy/sql/migrations/v2.4.0_0724_add_dreaming_lease_columns.sql b/deploy/sql/migrations/v2.4.0_0724_add_dreaming_lease_columns.sql new file mode 100644 index 000000000..9fc379ead --- /dev/null +++ b/deploy/sql/migrations/v2.4.0_0724_add_dreaming_lease_columns.sql @@ -0,0 +1,12 @@ +-- Add worker lease columns to the Dreaming audit table so the executor can +-- claim, renew, and release row-level leases with FOR UPDATE SKIP LOCKED. +SET search_path TO nexent; +BEGIN; + +ALTER TABLE nexent.memory_dreaming_audit_t + ADD COLUMN IF NOT EXISTS lock_owner VARCHAR(100); + +ALTER TABLE nexent.memory_dreaming_audit_t + ADD COLUMN IF NOT EXISTS lock_until TIMESTAMP; + +COMMIT; diff --git a/deploy/sql/migrations/v2.4.0_0727_add_memory_dreaming_schedule.sql b/deploy/sql/migrations/v2.4.0_0727_add_memory_dreaming_schedule.sql new file mode 100644 index 000000000..71f097970 --- /dev/null +++ b/deploy/sql/migrations/v2.4.0_0727_add_memory_dreaming_schedule.sql @@ -0,0 +1,30 @@ +CREATE TABLE IF NOT EXISTS nexent.memory_dreaming_schedule_t ( + schedule_id BIGSERIAL PRIMARY KEY, + tenant_id VARCHAR(100) NOT NULL, + user_id VARCHAR(100) NOT NULL, + agent_id VARCHAR(100) NOT NULL, + enabled BOOLEAN NOT NULL DEFAULT FALSE, + rule_type VARCHAR(20) NOT NULL DEFAULT 'CRON', + timezone VARCHAR(100) NOT NULL DEFAULT 'Asia/Shanghai', + start_at TIMESTAMP NOT NULL, + cron_expr VARCHAR(100), + interval_seconds INTEGER, + next_fire_at TIMESTAMP, + last_fire_at TIMESTAMP, + fire_count INTEGER NOT NULL DEFAULT 0, + 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', + CONSTRAINT ck_memory_dreaming_schedule_rule CHECK ( + (rule_type = 'CRON' AND cron_expr IS NOT NULL AND interval_seconds IS NULL) + OR + (rule_type = 'INTERVAL' AND cron_expr IS NULL AND interval_seconds >= 3600) + ) +); +CREATE UNIQUE INDEX IF NOT EXISTS uq_memory_dreaming_schedule_scope + ON nexent.memory_dreaming_schedule_t (tenant_id, user_id, agent_id); +CREATE INDEX IF NOT EXISTS idx_memory_dreaming_schedule_due + ON nexent.memory_dreaming_schedule_t (enabled, next_fire_at) + WHERE delete_flag = 'N'; diff --git a/frontend/app/[locale]/memory/DreamingPanel.tsx b/frontend/app/[locale]/memory/DreamingPanel.tsx new file mode 100644 index 000000000..8dccdc0ba --- /dev/null +++ b/frontend/app/[locale]/memory/DreamingPanel.tsx @@ -0,0 +1,588 @@ +"use client"; + +import React, { useCallback, useEffect, useMemo, useState } from "react"; +import { + Alert, + App, + Button, + Card, + Empty, + InputNumber, + List, + Progress, + Select, + Space, + Spin, + Switch, + Tag, + TimePicker, + Timeline, + Typography, +} from "antd"; +import { Brain, Clock, History, Play, RotateCcw, Save } from "lucide-react"; +import dayjs from "dayjs"; +import { useTranslation } from "react-i18next"; + +import { useAuthorizationContext } from "@/components/providers/AuthorizationProvider"; +import { + activateDreamingVersion, + DreamingAudit, + DreamingVersion, + fetchDreamingAudits, + fetchDreamingParameters, + fetchDreamingSchedule, + fetchDreamingVersions, + runDreaming, + saveDreamingSchedule, +} from "@/services/memoryService"; +import type { + DreamingParameters, + DreamingSchedule, +} from "@/services/memoryService"; +import { getTenantUsers, TenantUser } from "@/services/tenantService"; + +const phaseProgress: Record = { + light: 20, + rem: 45, + deep: 70, + compression: 90, +}; + +export function DreamingPanel() { + const { message } = App.useApp(); + const { t } = useTranslation("common"); + const { user, hasPermission } = useAuthorizationContext(); + const agentId = "__user__"; + const [tenantUsers, setTenantUsers] = useState([]); + const [targetUserId, setTargetUserId] = useState(); + const [audits, setAudits] = useState([]); + const [versions, setVersions] = useState([]); + const [parameters, setParameters] = useState(); + const [loading, setLoading] = useState(true); + const [triggering, setTriggering] = useState(false); + const [schedule, setSchedule] = useState(); + const [scheduleMode, setScheduleMode] = useState< + "daily" | "weekly" | "interval" + >("daily"); + const [scheduleTime, setScheduleTime] = useState("03:00"); + const [scheduleWeekday, setScheduleWeekday] = useState(1); + const [intervalHours, setIntervalHours] = useState(24); + const [savingSchedule, setSavingSchedule] = useState(false); + + const refresh = useCallback(async () => { + const target = + targetUserId && targetUserId !== user?.id ? targetUserId : undefined; + const [nextAudits, nextVersions, nextSchedule] = await Promise.all([ + fetchDreamingAudits(20, target), + fetchDreamingVersions(20, target), + fetchDreamingSchedule(target), + ]); + setAudits(nextAudits); + setVersions(nextVersions); + setSchedule(nextSchedule); + if (nextSchedule.rule_type === "INTERVAL") { + setScheduleMode("interval"); + setIntervalHours((nextSchedule.interval_seconds || 86400) / 3600); + } else { + const parts = (nextSchedule.cron_expr || "0 3 * * *").split(" "); + setScheduleTime( + `${parts[1].padStart(2, "0")}:${parts[0].padStart(2, "0")}` + ); + if (parts[4] !== "*") { + setScheduleMode("weekly"); + setScheduleWeekday(Number(parts[4])); + } else { + setScheduleMode("daily"); + } + } + }, [targetUserId, user?.id]); + + useEffect(() => { + if (user?.id && !targetUserId) setTargetUserId(user.id); + }, [targetUserId, user?.id]); + + useEffect(() => { + if (!user?.tenantId || !hasPermission("DREAMING:VIEW_TENANT")) { + setTenantUsers([]); + return; + } + getTenantUsers(user.tenantId) + .then(({ users }) => setTenantUsers(users)) + .catch(() => message.error(t("dreaming.error.loadUsers"))); + }, [hasPermission, message, t, user?.tenantId]); + + useEffect(() => { + fetchDreamingParameters() + .then((effectiveParameters) => { + setParameters(effectiveParameters); + }) + .catch(() => message.error(t("dreaming.error.loadAgents"))) + .finally(() => setLoading(false)); + }, [message, t]); + + useEffect(() => { + if (!agentId) return; + setLoading(true); + refresh() + .catch(() => message.error(t("dreaming.error.loadStatus"))) + .finally(() => setLoading(false)); + }, [agentId, message, refresh, t]); + + const activeRun = useMemo( + () => audits.find((run) => ["queued", "running"].includes(run.status)), + [audits] + ); + const latestRun = audits[0]; + const displayedSchedule: DreamingSchedule = schedule || { + agent_id: agentId || "", + enabled: false, + rule_type: "CRON", + timezone: "Asia/Shanghai", + cron_expr: "0 3 * * *", + interval_seconds: null, + fire_count: 0, + }; + const selectedIsSelf = !targetUserId || targetUserId === user?.id; + const canEditTarget = selectedIsSelf || hasPermission("DREAMING:EDIT_TENANT"); + const queueDelayed = + activeRun?.status === "queued" && + !!activeRun.started_at && + Date.now() - new Date(activeRun.started_at).getTime() > 60_000; + useEffect(() => { + if (!activeRun) return; + const timer = window.setInterval(() => refresh(), 2000); + return () => window.clearInterval(timer); + }, [activeRun, agentId, refresh]); + + const trigger = async () => { + if (!agentId) return; + setTriggering(true); + try { + await runDreaming(agentId, selectedIsSelf ? undefined : targetUserId); + message.success(t("dreaming.run.queued")); + await refresh(); + } catch { + message.error(t("dreaming.error.trigger")); + } finally { + setTriggering(false); + } + }; + + const saveSchedule = async () => { + if (!agentId || !schedule) return; + const [hour, minute] = scheduleTime.split(":").map(Number); + const timezone = + Intl.DateTimeFormat().resolvedOptions().timeZone || "Asia/Shanghai"; + setSavingSchedule(true); + try { + const saved = await saveDreamingSchedule({ + agent_id: agentId, + enabled: schedule.enabled, + rule_type: scheduleMode === "interval" ? "INTERVAL" : "CRON", + timezone, + start_at: new Date().toISOString(), + cron_expr: + scheduleMode === "interval" + ? null + : `${minute} ${hour} * * ${scheduleMode === "weekly" ? scheduleWeekday : "*"}`, + interval_seconds: + scheduleMode === "interval" ? Math.round(intervalHours * 3600) : null, + ...(selectedIsSelf ? {} : { target_user_id: targetUserId }), + }); + setSchedule(saved); + message.success(t("dreaming.schedule.saved")); + } catch { + message.error(t("dreaming.schedule.saveFailed")); + } finally { + setSavingSchedule(false); + } + }; + + const activate = async (version: DreamingVersion) => { + if (!agentId) return; + const activeVersion = versions.find((candidate) => candidate.is_active); + if (!activeVersion) return; + try { + await activateDreamingVersion( + agentId, + version.version_id, + activeVersion.version_id, + selectedIsSelf ? undefined : targetUserId + ); + message.success( + t("dreaming.version.switched", { version: version.version_no }) + ); + await refresh(); + } catch { + message.error(t("dreaming.error.activate")); + } + }; + + if (loading && !agentId) return ; + + return ( +
+ +
+
+ + + Dreaming + + + {t("dreaming.description")} + +
+ {parameters && ( + + + {t("dreaming.parameter.sourceLimit", { + count: parameters.source_limit, + })} + + + {t("dreaming.parameter.maxChars", { + count: parameters.long_term_max_chars, + })} + + + {t("dreaming.parameter.compressionRetries", { + count: parameters.compression_max_attempts, + })} + + + )} +
+
+ + {tenantUsers.length > 1 && ( + + {scheduleMode === "weekly" && ( +