feat(memory): implement Dreaming consolidation flow

backend/apps/config_app.py

@@ -41,6 +41,8 @@
 from apps.aidp_app import router as aidp_router
 from apps.cas_app import router as cas_router
 from apps.quota_app import tenant_quota_router, platform_quota_router
+from apps.memory_record_app import router as memory_record_router
+from apps.memory_dreaming_app import router as memory_dreaming_router
 from consts.const import IS_SPEED_MODE
 from services.prompt_template_service import sync_system_default_prompt_template
 
@@ -107,3 +109,5 @@ async def sync_default_prompt_template_on_startup():
 app.include_router(aidp_router)
 app.include_router(tenant_quota_router)
 app.include_router(platform_quota_router)
+app.include_router(memory_record_router)
+app.include_router(memory_dreaming_router)

backend/apps/memory_dreaming_app.py

@@ -0,0 +1,54 @@
+"""Manual Dreaming run and audit endpoints."""
+
+from http import HTTPStatus
+from typing import Annotated, Optional
+
+from fastapi import APIRouter, Header, HTTPException, Query
+from pydantic import BaseModel, Field
+
+from services.memory_dreaming_service import (
+    DreamingRunError,
+    get_memory_dreaming_service,
+)
+from utils.auth_utils import get_current_user_id
+
+router = APIRouter(prefix="/memory/dreaming", tags=["memory-dreaming"])
+
+
+class DreamingRunRequest(BaseModel):
+    agent_id: str = Field(..., min_length=1)
+
+
+@router.post("/run")
+def run_dreaming(
+    payload: DreamingRunRequest,
+    authorization: Annotated[Optional[str], Header()] = None,
+):
+    user_id, tenant_id = get_current_user_id(authorization)
+    try:
+        return get_memory_dreaming_service().run(
+            tenant_id=tenant_id,
+            user_id=user_id,
+            agent_id=payload.agent_id,
+        )
+    except DreamingRunError as exc:
+        raise HTTPException(
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(exc)
+        ) from exc
+
+
+@router.get("/audit")
+def list_dreaming_audits(
+    authorization: Annotated[Optional[str], Header()] = None,
+    agent_id: Annotated[Optional[str], Query()] = None,
+    run_id: Annotated[Optional[int], Query(ge=1)] = None,
+    limit: Annotated[int, Query(ge=1, le=500)] = 100,
+):
+    user_id, tenant_id = get_current_user_id(authorization)
+    return get_memory_dreaming_service().list_audits(
+        tenant_id,
+        user_id,
+        agent_id=agent_id,
+        run_id=run_id,
+        limit=limit,
+    )

backend/database/db_models.py

@@ -760,6 +760,111 @@ class MemoryUserConfig(TableBase):
     config_value = Column(String(10000), doc="the value of the config")
 
 
+class MemoryRecord(TableBase):
+    """Authoritative tenant/user/agent memory row."""
+
+    __tablename__ = "memory_records_t"
+    __table_args__ = (
+        Index("idx_memory_records_tenant", "tenant_id"),
+        Index("idx_memory_records_user", "tenant_id", "user_id"),
+        Index("idx_memory_records_agent", "tenant_id", "user_id", "agent_id", "conversation_id"),
+        Index("idx_memory_records_idempotency", "tenant_id", "idempotency_key"),
+        Index("idx_memory_records_status", "tenant_id", "user_id", "layer", "status"),
+        {"schema": SCHEMA},
+    )
+
+    memory_id = Column(Integer, primary_key=True, nullable=False, autoincrement=True)
+    tenant_id = Column(String(100), nullable=False)
+    user_id = Column(String(100), nullable=False)
+    agent_id = Column(String(100))
+    conversation_id = Column(String(100))
+    layer = Column(String(30), nullable=False)
+    memory_type = Column(String(30))
+    status = Column(String(30), nullable=False, default="active")
+    content = Column(Text, nullable=False)
+    concept_tags = Column(ARRAY(Text))
+    es_index_name = Column(String(255))
+    idempotency_key = Column(String(128), nullable=False)
+    recall_count = Column(Integer, nullable=False, default=0)
+    daily_count = Column(Integer, nullable=False, default=0)
+    grounded_count = Column(Integer, nullable=False, default=0)
+    last_recalled_at = Column(TIMESTAMP(timezone=False))
+    query_hashes = Column(ARRAY(Text))
+    recall_days = Column(ARRAY(Text))
+    light_hits = Column(Integer, nullable=False, default=0)
+    rem_hits = Column(Integer, nullable=False, default=0)
+    last_light_at = Column(TIMESTAMP(timezone=False))
+    last_rem_at = Column(TIMESTAMP(timezone=False))
+
+
+class MemoryRetrievalHit(TableBase):
+    """Append-only recall evidence consumed by Dreaming."""
+
+    __tablename__ = "memory_retrieval_hits_t"
+    __table_args__ = (
+        Index("idx_memory_retrieval_hits_memory", "memory_id", "occurred_at"),
+        Index(
+            "idx_memory_retrieval_hits_tenant_user_agent",
+            "tenant_id",
+            "user_id",
+            "agent_id",
+            "day",
+        ),
+        {"schema": SCHEMA},
+    )
+
+    hit_id = Column(Integer, primary_key=True, nullable=False, autoincrement=True)
+    tenant_id = Column(String(100))
+    user_id = Column(String(100))
+    agent_id = Column(String(100))
+    conversation_id = Column(String(100))
+    memory_id = Column(Integer)
+    query_text = Column(Text)
+    query_hash = Column(String(128))
+    retrieval_score = Column(Numeric(38, 18))
+    source = Column(String(100), nullable=False, default="nexent")
+    occurred_at = Column(TIMESTAMP(timezone=False), nullable=False, server_default=func.now())
+    day = Column(String(100))
+    grounded = Column(Boolean, nullable=False, default=False)
+
+
+class MemoryDreamingAudit(TableBase):
+    """One durable audit row per manual Dreaming run."""
+
+    __tablename__ = "memory_dreaming_audit_t"
+    __table_args__ = (
+        Index(
+            "idx_memory_dreaming_audit_scope",
+            "tenant_id",
+            "user_id",
+            "agent_id",
+            "started_at",
+        ),
+        {"schema": SCHEMA},
+    )
+
+    run_id = Column(
+        BigInteger,
+        Sequence("memory_dreaming_audit_t_run_id_seq", schema=SCHEMA),
+        primary_key=True,
+        nullable=False,
+    )
+    tenant_id = Column(String(100), nullable=False)
+    user_id = Column(String(100), nullable=False)
+    agent_id = Column(String(100), nullable=False)
+    trigger_source = Column(String(30), nullable=False, default="manual")
+    status = Column(String(30), nullable=False, default="running")
+    current_phase = Column(String(30))
+    started_at = Column(TIMESTAMP(timezone=False), nullable=False, server_default=func.now())
+    finished_at = Column(TIMESTAMP(timezone=False))
+    light_count = Column(Integer, nullable=False, default=0)
+    rem_count = Column(Integer, nullable=False, default=0)
+    promoted_count = Column(Integer, nullable=False, default=0)
+    deferred_count = Column(Integer, nullable=False, default=0)
+    result_json = Column(JSONB)
+    error = Column(Text)
+
+
 class McpRecord(TableBase):
     """
     MCP (Model Context Protocol) records table
@@ -1093,32 +1198,6 @@ class SkillRepository(TableBase):
                     doc="Listing status: not_shared / pending_review / rejected / shared")
 
 
-class SkillRepository(TableBase):
-    """
-    Skill repository (marketplace) table. Frozen snapshot of a shared skill for installation.
-    """
-    __tablename__ = "ag_skill_repository_t"
-    __table_args__ = {"schema": SCHEMA}
-
-    skill_repository_id = Column(BigInteger, Sequence("ag_skill_repository_t_skill_repository_id_seq", schema=SCHEMA),
-                                 primary_key=True, nullable=False, doc="Skill repository listing ID, unique primary key")
-    publisher_tenant_id = Column(String(100), nullable=False, doc=_PUBLISHER_TENANT_ID_DOC)
-    publisher_user_id = Column(String(100), nullable=False, doc=_PUBLISHER_USER_ID_DOC)
-    skill_id = Column(Integer, nullable=False, doc="Source skill ID from ag_skill_info_t")
-    name = Column(String(100), nullable=False, doc="Skill name for display and search")
-    description = Column(Text, doc="Skill description")
-    source = Column(String(30), doc="Skill source")
-    submitted_by = Column(String(100), doc="Submitter email when listing enters pending_review")
-    category_id = Column(Integer, doc="Optional marketplace category ID")
-    tags = Column(ARRAY(Text), doc="Marketplace tags")
-    icon = Column(String(100), doc="Marketplace card icon (emoji or URL)")
-    downloads = Column(Integer, default=0, doc="Marketplace install count for card display")
-    skill_info_json = Column(JSONB, nullable=False, doc="Frozen skill metadata snapshot")
-    skill_zip_base64 = Column(Text, nullable=False, doc="Frozen skill ZIP payload encoded as base64")
-    status = Column(String(30), default="not_shared",
-                    doc="Listing status: not_shared / pending_review / rejected / shared")
-
-
 class UserTokenInfo(TableBase):
     """
     User token (AK/SK) information table

backend/database/memory_dreaming_db.py

@@ -0,0 +1,132 @@
+"""Persistence and PostgreSQL advisory locking for manual Dreaming runs."""
+
+from __future__ import annotations
+
+import hashlib
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from typing import Any, Dict, Iterator, List, Optional
+
+from sqlalchemy import text
+
+from .client import get_db_session
+from .db_models import MemoryDreamingAudit
+
+
+def advisory_lock_key(tenant_id: str, user_id: str, agent_id: str) -> int:
+    digest = hashlib.sha256(
+        f"{tenant_id}:{user_id}:{agent_id}".encode("utf-8")
+    ).digest()
+    return int.from_bytes(digest[:8], "big", signed=True)
+
+
+@contextmanager
+def try_scope_lock(tenant_id: str, user_id: str, agent_id: str) -> Iterator[bool]:
+    """Hold a transaction-scoped advisory lock for the context lifetime."""
+    with get_db_session() as session:
+        acquired = bool(
+            session.execute(
+                text("SELECT pg_try_advisory_xact_lock(:lock_key)"),
+                {"lock_key": advisory_lock_key(tenant_id, user_id, agent_id)},
+            ).scalar()
+        )
+        try:
+            yield acquired
+            session.commit()
+        except Exception:
+            session.rollback()
+            raise
+
+
+def create_audit(tenant_id: str, user_id: str, agent_id: str) -> int:
+    with get_db_session() as session:
+        row = MemoryDreamingAudit(
+            tenant_id=tenant_id,
+            user_id=user_id,
+            agent_id=agent_id,
+            trigger_source="manual",
+            status="running",
+            current_phase="light",
+        )
+        session.add(row)
+        session.commit()
+        return int(row.run_id)
+
+
+def update_audit(run_id: int, values: Dict[str, Any]) -> bool:
+    allowed = {
+        "status",
+        "current_phase",
+        "finished_at",
+        "light_count",
+        "rem_count",
+        "promoted_count",
+        "deferred_count",
+        "result_json",
+        "error",
+    }
+    with get_db_session() as session:
+        row = (
+            session.query(MemoryDreamingAudit)
+            .filter(MemoryDreamingAudit.run_id == run_id)
+            .first()
+        )
+        if row is None:
+            return False
+        for key, value in values.items():
+            if key in allowed:
+                setattr(row, key, value)
+        session.commit()
+        return True
+
+
+def finish_audit(run_id: int, *, status: str, **values: Any) -> bool:
+    payload = {
+        **values,
+        "status": status,
+        "finished_at": datetime.now(timezone.utc).replace(tzinfo=None),
+    }
+    if status != "failed":
+        payload["current_phase"] = None
+    return update_audit(run_id, payload)
+
+
+def list_audits(
+    tenant_id: str,
+    user_id: str,
+    *,
+    agent_id: Optional[str] = None,
+    run_id: Optional[int] = None,
+    limit: int = 100,
+) -> List[Dict[str, Any]]:
+    with get_db_session() as session:
+        query = session.query(MemoryDreamingAudit).filter(
+            MemoryDreamingAudit.tenant_id == tenant_id,
+            MemoryDreamingAudit.user_id == user_id,
+            MemoryDreamingAudit.delete_flag == "N",
+        )
+        if agent_id is not None:
+            query = query.filter(MemoryDreamingAudit.agent_id == agent_id)
+        if run_id is not None:
+            query = query.filter(MemoryDreamingAudit.run_id == run_id)
+        rows = query.order_by(MemoryDreamingAudit.run_id.desc()).limit(limit).all()
+        return [
+            {
+                "run_id": row.run_id,
+                "tenant_id": row.tenant_id,
+                "user_id": row.user_id,
+                "agent_id": row.agent_id,
+                "trigger_source": row.trigger_source,
+                "status": row.status,
+                "current_phase": row.current_phase,
+                "started_at": row.started_at.isoformat() if row.started_at else None,
+                "finished_at": row.finished_at.isoformat() if row.finished_at else None,
+                "light_count": row.light_count,
+                "rem_count": row.rem_count,
+                "promoted_count": row.promoted_count,
+                "deferred_count": row.deferred_count,
+                "result": row.result_json,
+                "error": row.error,
+            }
+            for row in rows
+        ]