feat: 实现用户画像、占卜历史与后端用户管理模块

backend/src/v1/agent/repository.py

@@ -335,6 +335,59 @@ class AgentRepository:
             return None
         return str(latest_id)
 
+    async def get_latest_assistant_messages_by_user_sessions(
+        self,
+        *,
+        user_id: str,
+        visibility_mask: int | None = None,
+        session_limit: int = 50,
+    ) -> list[dict[str, object]]:
+        try:
+            user_uuid = UUID(user_id)
+        except ValueError as exc:
+            raise ApiProblemError(
+                status_code=422,
+                code="AGENT_USER_ID_INVALID",
+                detail="Invalid user_id",
+            ) from exc
+
+        safe_limit = max(int(session_limit), 1)
+        session_stmt = (
+            select(AgentChatSession.id)
+            .where(AgentChatSession.user_id == user_uuid)
+            .where(AgentChatSession.deleted_at.is_(None))
+            .order_by(AgentChatSession.last_activity_at.desc())
+            .limit(safe_limit)
+        )
+        session_ids = (await self._session.execute(session_stmt)).scalars().all()
+        if not session_ids:
+            return []
+
+        snapshots: list[dict[str, object]] = []
+        for session_id in session_ids:
+            message_stmt = (
+                select(AgentChatMessage)
+                .where(AgentChatMessage.session_id == session_id)
+                .where(AgentChatMessage.deleted_at.is_(None))
+                .where(AgentChatMessage.role == AgentChatMessageRole.ASSISTANT)
+                .order_by(AgentChatMessage.created_at.desc())
+                .limit(1)
+            )
+            message_stmt = self._apply_visibility_filter(
+                stmt=message_stmt,
+                visibility_mask=visibility_mask,
+            )
+            message = (await self._session.execute(message_stmt)).scalar_one_or_none()
+            if message is None:
+                continue
+            snapshots.append(await self._to_snapshot_message(message))
+
+        snapshots.sort(
+            key=lambda item: str(item.get("timestamp") or ""),
+            reverse=True,
+        )
+        return snapshots
+
     async def get_system_agent_config(
         self, *, agent_type: str
     ) -> dict[str, object] | None:

backend/src/v1/agent/schemas.py

@@ -7,7 +7,7 @@ from uuid import UUID
 
 from pydantic import BaseModel, ConfigDict, Field
 
-from schemas.agent.ui_schema import UiSchemaRenderer
+from schemas.domain.divination import DerivedDivinationData
 
 
 class AgentRepositoryLike(Protocol):
@@ -31,6 +31,14 @@ class AgentRepositoryLike(Protocol):
 
     async def get_latest_session_id_for_user(self, *, user_id: str) -> str | None: ...
 
+    async def get_latest_assistant_messages_by_user_sessions(
+        self,
+        *,
+        user_id: str,
+        visibility_mask: int | None = None,
+        session_limit: int = 50,
+    ) -> list[dict[str, object]]: ...
+
     async def persist_user_message(
         self,
         *,
@@ -187,13 +195,31 @@ class HistoryMessage(BaseModel):
         default_factory=list,
         description="Temporary signed URLs for user-attached images",
     )
-    ui_schema: UiSchemaRenderer | None = Field(
+
+    agent_output: HistoryAgentOutput | None = Field(
         default=None,
-        description="Compiled UI schema from worker ui_hints for frontend rendering",
+        description="Structured assistant output for history replay",
     )
     timestamp: str = Field(description="Message creation timestamp in ISO-8601 format")
 
 
+class HistoryAgentOutput(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    status: Literal["success", "failed"] | None = None
+    sign_level: Literal["上上签", "中上签", "中下签", "下下签"] | None = None
+    summary: str | None = None
+    conclusion: list[str] = Field(default_factory=list)
+    focus_points: list[str] = Field(default_factory=list)
+    advice: list[str] = Field(default_factory=list)
+    keywords: list[str] = Field(default_factory=list)
+    answer: str | None = None
+    key_points: list[str] = Field(default_factory=list)
+    result_type: str | None = None
+    suggested_actions: list[str] = Field(default_factory=list)
+    divination_derived: DerivedDivinationData | None = None
+
+
 class HistorySnapshotResponse(BaseModel):
     """Response schema for GET /api/v1/agent/history"""
 

backend/src/v1/agent/service.py

@@ -641,23 +641,37 @@ class AgentService:
         thread_id: str | None,
         before: date | None,
     ) -> HistorySnapshotResponse:
-        target_thread_id = thread_id
-        if target_thread_id is None:
-            target_thread_id = await self._repository.get_latest_session_id_for_user(
-                user_id=str(current_user.id)
+        from schemas.domain.chat_message import AgentChatMessage
+        from v1.agent.utils import convert_message_to_history
+        from v1.agent.schemas import HistoryMessage
+
+        if thread_id is not None:
+            return await self.get_history_snapshot(
+                thread_id=thread_id,
+                before=before,
+                current_user=current_user,
             )
-        if target_thread_id is None:
-            return HistorySnapshotResponse(
-                scope="history_day",
-                threadId=None,
-                day=None,
-                hasMore=False,
-                messages=[],
+
+        raw_messages = (
+            await self._repository.get_latest_assistant_messages_by_user_sessions(
+                user_id=str(current_user.id),
+                visibility_mask=bit_mask(bit=int(SystemVisibilityBit.UI_HISTORY)),
+                session_limit=50,
             )
-        return await self.get_history_snapshot(
-            thread_id=target_thread_id,
-            before=before,
-            current_user=current_user,
+        )
+
+        messages: list[HistoryMessage] = []
+        for msg_dict in raw_messages:
+            msg = AgentChatMessage.model_validate(msg_dict)
+            converted = convert_message_to_history(msg)
+            messages.append(HistoryMessage.model_validate(converted))
+
+        return HistorySnapshotResponse(
+            scope="history_sessions_latest_assistant",
+            threadId=None,
+            day=None,
+            hasMore=False,
+            messages=messages,
         )
 
     def _validate_binary_signed_url(

backend/src/v1/agent/utils.py

@@ -7,7 +7,7 @@
 from collections.abc import Callable
 from typing import Any
 
-from core.agentscope.runtime.ui_compiler import compile as compile_ui_hints
+from schemas.agent.runtime_models import AgentOutput
 from schemas.domain.chat_message import (
     AgentChatMessage,
     AgentChatMessageMetadata,
@@ -29,20 +29,20 @@ def convert_message_to_history(
 
     转换规则：
     - role=user: 读取 metadata.user_message_attachments，转换为 attachments[]
-    - role=assistant: 读取 metadata.agent_output.ui_hints，编译成 ui_schema
+    - role=assistant: 读取 metadata.agent_output，输出受控 agent_output
     """
     role = message.role
     content = message.content
     metadata = message.metadata
 
     attachments: list[dict[str, str]] = []
-    ui_schema: dict[str, Any] | None = None
+    agent_output: dict[str, Any] | None = None
 
     if role == "user":
         attachments = _convert_user_attachments(metadata, get_signed_url_fn)
 
     elif role == "assistant":
-        ui_schema = _compile_worker_ui_hints(metadata)
+        agent_output = _extract_worker_agent_output(metadata)
 
     result: dict[str, Any] = {
         "id": str(message.id),
@@ -55,8 +55,8 @@ def convert_message_to_history(
     if attachments:
         result["attachments"] = attachments
 
-    if ui_schema:
-        result["ui_schema"] = ui_schema
+    if agent_output:
+        result["agent_output"] = agent_output
 
     return result
 
@@ -93,10 +93,10 @@ def _convert_user_attachments(
     return signed_attachments
 
 
-def _compile_worker_ui_hints(
+def _extract_worker_agent_output(
     metadata: AgentChatMessageMetadata | dict[str, Any] | None,
 ) -> dict[str, Any] | None:
-    """编译 assistant 消息的 agent ui_hints"""
+    """提取 assistant 消息的结构化 agent_output。"""
     if not metadata:
         return None
 
@@ -106,29 +106,52 @@ def _compile_worker_ui_hints(
         agent_output_data = metadata.get("agent_output")
         if not agent_output_data:
             return None
-        if isinstance(agent_output_data, dict):
-            raw_ui_schema = agent_output_data.get("ui_schema")
-            if isinstance(raw_ui_schema, dict):
-                return raw_ui_schema
-        from schemas.agent.runtime_models import AgentOutput
-
         try:
             agent_output = AgentOutput.model_validate(agent_output_data)
         except Exception:
-            return None
+            normalized_payload = _normalize_agent_output_payload(agent_output_data)
+            try:
+                agent_output = AgentOutput.model_validate(normalized_payload)
+            except Exception:
+                return None
 
     if not agent_output:
         return None
 
-    ui_hints = agent_output.ui_hints
-    if not ui_hints:
-        return None
+    payload = agent_output.model_dump(mode="json", by_alias=True, exclude_none=True)
+    payload.pop("ui_hints", None)
+    return payload or None
 
-    try:
-        compiled = compile_ui_hints(ui_hints)
-        return compiled
-    except Exception:
+
+def _normalize_agent_output_payload(agent_output_data: Any) -> dict[str, Any] | None:
+    if not isinstance(agent_output_data, dict):
         return None
+    normalized = dict(agent_output_data)
+    derived = normalized.get("divination_derived")
+    if isinstance(derived, dict):
+        normalized["divination_derived"] = _normalize_divination_derived(derived)
+    return normalized
+
+
+def _normalize_divination_derived(value: Any) -> Any:
+    if isinstance(value, dict):
+        result: dict[str, Any] = {}
+        for key, item in value.items():
+            normalized_key = _snake_to_camel(key)
+            result[normalized_key] = _normalize_divination_derived(item)
+        return result
+    if isinstance(value, list):
+        return [_normalize_divination_derived(item) for item in value]
+    return value
+
+
+def _snake_to_camel(value: str) -> str:
+    if "_" not in value:
+        return value
+    parts = value.split("_")
+    if not parts:
+        return value
+    return parts[0] + "".join(part[:1].upper() + part[1:] for part in parts[1:])
 
 
 def mime_to_suffix(mime_type: str) -> str: