diff --git a/backend/package/yuxi/agents/middlewares/subagent_task.py b/backend/package/yuxi/agents/middlewares/subagent_task.py
index 4d225cd3..6b489ff4 100644
--- a/backend/package/yuxi/agents/middlewares/subagent_task.py
+++ b/backend/package/yuxi/agents/middlewares/subagent_task.py
@@ -17,14 +17,17 @@ from langgraph.types import Command
 from yuxi.agents.context import build_agent_input_context
 from yuxi.repositories.agent_repository import SUB_AGENT_BACKEND_ID, AgentRepository
 from yuxi.repositories.agent_run_repository import AgentRunRepository
+from yuxi.repositories.conversation_repository import ConversationRepository
 from yuxi.repositories.user_repository import UserRepository
 from yuxi.storage.postgres.manager import pg_manager
 from yuxi.storage.postgres.models_business import Agent
+from yuxi.utils import logger
 from yuxi.utils.datetime_utils import utc_isoformat
 from yuxi.utils.subagent_thread_utils import make_child_thread_id
 
 _CHILD_STATE_INHERIT_KEYS: frozenset[str] = frozenset()
 _TERMINAL_RUN_STATUSES = {"completed", "failed", "cancelled", "interrupted"}
+_MAX_SPAWN_DEPTH = 3
 
 TASK_SYSTEM_PROMPT = """## `task`（子智能体任务工具）
 
@@ -372,6 +375,19 @@ class YuxiSubAgentMiddleware(AgentMiddleware[Any, ContextT, ResponseT]):
             if not backend or agent.backend_id != SUB_AGENT_BACKEND_ID:
                 return f"无法调用子智能体 {subagent_type}：后端配置无效"
 
+            # 嵌套深度限制：查询当前会话的 spawn_depth，超过上限则拒绝
+            parent_spawn_depth: int | None = None
+            try:
+                async with pg_manager.get_async_session_context() as db:
+                    conv_repo = ConversationRepository(db)
+                    current_conv = await conv_repo.get_conversation_by_thread_id(parent_thread_id)
+                    parent_spawn_depth = current_conv.spawn_depth if current_conv else None
+                    current_depth = parent_spawn_depth if parent_spawn_depth is not None else 0
+                if current_depth >= _MAX_SPAWN_DEPTH:
+                    return f"子智能体嵌套深度已达上限（{_MAX_SPAWN_DEPTH} 层），无法继续派生子智能体"
+            except Exception:
+                logger.warning("Failed to check spawn depth, allowing subagent invocation")
+
             child_thread_id, continuing = _new_child_thread_id(
                 thread_id,
                 parent_thread_id=parent_thread_id,
@@ -463,6 +479,20 @@ class YuxiSubAgentMiddleware(AgentMiddleware[Any, ContextT, ResponseT]):
                 )
                 return _failed_tool_response(exc, runtime.tool_call_id, _with_run_payload(subagent_run, failed_run))
             completed_run = await self._set_subagent_run_status(run.id, "completed")
+
+            # 更新子会话的父子关系（复用之前查询的 parent_spawn_depth，避免重复开 session 查询父会话）
+            try:
+                async with pg_manager.get_async_session_context() as db:
+                    conv_repo = ConversationRepository(db)
+                    child_conv = await conv_repo.get_conversation_by_thread_id(child_thread_id)
+                    if child_conv:
+                        depth = (parent_spawn_depth + 1) if parent_spawn_depth is not None else 1
+                        child_conv.parent_thread_id = parent_thread_id
+                        child_conv.spawn_depth = depth
+                        await db.commit()
+            except Exception:
+                logger.warning(f"Failed to update parent relationship for child thread {child_thread_id}")
+
             return _completed_tool_response(
                 result, runtime.tool_call_id, _with_run_payload(subagent_run, completed_run)
             )
diff --git a/backend/package/yuxi/agents/toolkits/buildin/__init__.py b/backend/package/yuxi/agents/toolkits/buildin/__init__.py
index 88903f78..072d2d31 100644
--- a/backend/package/yuxi/agents/toolkits/buildin/__init__.py
+++ b/backend/package/yuxi/agents/toolkits/buildin/__init__.py
@@ -2,8 +2,22 @@
 from .install_skill import install_skill
 from .tools import ask_user_question, present_artifacts
 
+# 会话通信工具仅在非 LITE 模式下注册
+from .session_tools import _LITE_MODE as _session_lite_mode
+
+if not _session_lite_mode:
+    from .session_tools import get_agent_progress, get_session_history, list_sessions, send_to_session
+
 __all__ = [
     "ask_user_question",
     "install_skill",
     "present_artifacts",
 ]
+
+if not _session_lite_mode:
+    __all__.extend([
+        "get_agent_progress",
+        "get_session_history",
+        "list_sessions",
+        "send_to_session",
+    ])
diff --git a/backend/package/yuxi/agents/toolkits/buildin/session_tools.py b/backend/package/yuxi/agents/toolkits/buildin/session_tools.py
new file mode 100644
index 00000000..88a1db88
--- /dev/null
+++ b/backend/package/yuxi/agents/toolkits/buildin/session_tools.py
@@ -0,0 +1,318 @@
+"""会话通信工具集 — Agent 间会话通信能力"""
+
+import os
+
+from langgraph.prebuilt.tool_node import ToolRuntime
+from pydantic import BaseModel, Field
+
+from yuxi.agents.toolkits.registry import tool
+from yuxi.utils import logger
+
+_LITE_MODE = os.environ.get("LITE_MODE", "").lower() in ("true", "1")
+
+# ── 常量 ──────────────────────────────────────────────────
+
+_HISTORY_CONTENT_MAX_CHARS = 4000
+_HISTORY_BYTES_MAX = 80 * 1024
+
+
+# ── 辅助函数 ──────────────────────────────────────────────
+
+
+def _get_runtime_uid(runtime: ToolRuntime) -> str:
+    """从 ToolRuntime.context 获取当前用户 uid。"""
+    uid = getattr(runtime.context, "uid", None)
+    if not uid:
+        raise ValueError("当前运行时缺少 uid")
+    return str(uid)
+
+
+def _get_runtime_thread_id(runtime: ToolRuntime) -> str:
+    """从 ToolRuntime.context 获取当前线程 ID。
+
+    注意：子智能体的 file_thread_id 继承自父会话，不能用于标识当前会话。
+    会话通信场景应使用 thread_id（当前智能体自身的线程）。
+    """
+    thread_id = getattr(runtime.context, "thread_id", None)
+    if not thread_id:
+        raise ValueError("当前运行时缺少 thread_id")
+    return str(thread_id)
+
+
+# ── 权限校验 ──────────────────────────────────────────────
+
+
+async def _check_session_access(uid: str, thread_id: str) -> None:
+    """
+    检查用户是否有权访问指定会话（tree 级可见性）。
+    1. 会话的 uid 与当前用户一致 → 允许
+    2. 沿 parent_thread_id 链上溯，任一祖先属于当前用户 → 允许
+    3. 否则 → 拒绝
+    """
+    from yuxi.repositories.conversation_repository import ConversationRepository
+    from yuxi.storage.postgres.manager import pg_manager
+
+    async with pg_manager.get_async_session_context() as db:
+        repo = ConversationRepository(db)
+        visited: set[str] = set()
+        current_tid = thread_id
+
+        while current_tid and current_tid not in visited:
+            visited.add(current_tid)
+            conv = await repo.get_conversation_by_thread_id(current_tid)
+            if not conv:
+                raise ValueError(f"会话不存在: {thread_id}")
+            if conv.uid == uid:
+                return
+            parent_tid = conv.parent_thread_id
+            if not parent_tid:
+                break
+            current_tid = parent_tid
+
+    raise PermissionError(f"无权访问会话: {thread_id}")
+
+
+# ── LITE 模式下不注册会话通信工具 ─────────────────────────
+
+if _LITE_MODE:
+    logger.info("LITE_MODE enabled, session communication tools not registered")
+else:
+
+    # ── list_sessions ─────────────────────────────────────────
+
+    class ListSessionsInput(BaseModel):
+        status: str | None = Field(default=None, description="按状态过滤: active / archived")
+        agent_id: str | None = Field(default=None, description="按智能体 ID 过滤")
+        limit: int = Field(default=20, ge=1, le=50, description="返回数量上限")
+
+    @tool(
+        category="buildin",
+        tags=["会话"],
+        display_name="列出会话",
+        args_schema=ListSessionsInput,
+    )
+    async def list_sessions(
+        status: str | None = None,
+        agent_id: str | None = None,
+        limit: int = 20,
+        runtime: ToolRuntime = None,
+    ) -> str:
+        """
+        列出当前用户可见的对话会话，包括自己创建的和子智能体的会话。
+        返回每个会话的 thread_id、agent_id、状态、标题、最近更新时间。
+        """
+        from yuxi.repositories.conversation_repository import ConversationRepository
+        from yuxi.storage.postgres.manager import pg_manager
+
+        uid = _get_runtime_uid(runtime)
+
+        async with pg_manager.get_async_session_context() as db:
+            repo = ConversationRepository(db)
+            conversations = await repo.list_conversations(
+                uid=uid,
+                agent_id=agent_id,
+                status=status or "active",
+                limit=limit,
+            )
+
+        if not conversations:
+            return "当前无可见的会话"
+
+        lines = []
+        for conv in conversations:
+            parent_marker = " [子智能体]" if conv.parent_thread_id else ""
+            lines.append(
+                f"- thread_id: {conv.thread_id} | agent: {conv.agent_id} | "
+                f"状态: {conv.status} | 标题: {conv.title or '无'}{parent_marker} | "
+                f"更新: {conv.updated_at}"
+            )
+        return "\n".join(lines)
+
+    # ── get_session_history ───────────────────────────────────
+
+    class GetSessionHistoryInput(BaseModel):
+        thread_id: str = Field(description="目标会话线程 ID")
+        limit: int = Field(default=10, ge=1, le=50, description="返回消息数量")
+        include_tools: bool = Field(default=False, description="是否包含工具调用消息")
+
+    @tool(
+        category="buildin",
+        tags=["会话"],
+        display_name="查看会话历史",
+        args_schema=GetSessionHistoryInput,
+    )
+    async def get_session_history(
+        thread_id: str,
+        limit: int = 10,
+        include_tools: bool = False,
+        runtime: ToolRuntime = None,
+    ) -> str:
+        """
+        获取指定会话的最近消息。
+        仅可查看自己或子智能体的会话历史。
+        超长内容会被截断，工具调用消息默认不包含。
+        """
+        from yuxi.repositories.conversation_repository import ConversationRepository
+        from yuxi.storage.postgres.manager import pg_manager
+
+        uid = _get_runtime_uid(runtime)
+        await _check_session_access(uid, thread_id)
+
+        async with pg_manager.get_async_session_context() as db:
+            repo = ConversationRepository(db)
+            # get_messages_by_thread_id 按 created_at.asc() 排序 + limit 返回最旧消息，
+            # 但我们需要最近消息，所以先降序查再反转。
+            conversation = await repo.get_conversation_by_thread_id(thread_id)
+            if not conversation:
+                return f"会话 {thread_id} 暂无消息"
+            from sqlalchemy import select
+
+            from yuxi.storage.postgres.models_business import Message
+
+            result = await db.execute(
+                select(Message)
+                .where(Message.conversation_id == conversation.id)
+                .order_by(Message.created_at.desc())
+                .limit(limit)
+            )
+            messages = list(reversed(result.scalars().unique().all()))
+
+        if not messages:
+            return f"会话 {thread_id} 暂无消息"
+
+        total_bytes = 0
+        lines = []
+        truncated = False
+
+        for msg in messages:
+            # 过滤工具消息
+            if not include_tools and msg.message_type in ("tool_call", "tool_result"):
+                continue
+
+            content = msg.content or ""
+            # 截断超长内容
+            if len(content) > _HISTORY_CONTENT_MAX_CHARS:
+                content = content[:_HISTORY_CONTENT_MAX_CHARS] + "...[已截断]"
+                truncated = True
+
+            line = f"[{msg.role}] {content}"
+            line_bytes = len(line.encode("utf-8"))
+            if total_bytes + line_bytes > _HISTORY_BYTES_MAX:
+                truncated = True
+                break
+            total_bytes += line_bytes
+            lines.append(line)
+
+        suffix = "\n(部分内容已截断)" if truncated else ""
+        return "\n".join(lines) + suffix
+
+    # ── send_to_session ───────────────────────────────────────
+
+    class SendToSessionInput(BaseModel):
+        thread_id: str = Field(description="目标会话线程 ID")
+        message: str = Field(description="要发送的消息内容")
+
+    @tool(
+        category="buildin",
+        tags=["会话"],
+        display_name="发送会话消息",
+        args_schema=SendToSessionInput,
+    )
+    async def send_to_session(
+        thread_id: str,
+        message: str,
+        runtime: ToolRuntime = None,
+    ) -> str:
+        """
+        向指定会话发送消息。
+        消息会以 user 角色注入目标会话的历史记录。
+        如果目标会话正在运行，消息将作为下一轮输入；否则仅写入历史。
+        不能向自己所在的会话发送消息（避免循环）。
+        """
+        from yuxi.repositories.conversation_repository import ConversationRepository
+        from yuxi.storage.postgres.manager import pg_manager
+
+        uid = _get_runtime_uid(runtime)
+        current_thread_id = _get_runtime_thread_id(runtime)
+
+        # 防止向自身会话发送消息
+        if thread_id == current_thread_id:
+            return "不能向当前所在的会话发送消息"
+
+        await _check_session_access(uid, thread_id)
+
+        # 标注消息来源
+        annotated_message = f"[来自会话 {current_thread_id}] {message}"
+
+        async with pg_manager.get_async_session_context() as db:
+            repo = ConversationRepository(db)
+            await repo.add_message_by_thread_id(
+                thread_id=thread_id,
+                role="user",
+                content=annotated_message,
+                message_type="text",
+                extra_metadata={
+                    "source_thread_id": current_thread_id,
+                    "source_type": "inter_session",
+                },
+            )
+
+        return f"消息已发送到会话 {thread_id}"
+
+    # ── get_agent_progress ────────────────────────────────────
+
+    class GetAgentProgressInput(BaseModel):
+        thread_id: str = Field(description="会话线程 ID")
+
+    @tool(
+        category="buildin",
+        tags=["会话"],
+        display_name="查看智能体进度",
+        args_schema=GetAgentProgressInput,
+    )
+    async def get_agent_progress(
+        thread_id: str,
+        runtime: ToolRuntime = None,
+    ) -> str:
+        """
+        查看指定智能体的运行进度。
+        返回运行状态、开始时间、运行类型等信息。
+        """
+        from sqlalchemy import select
+
+        from yuxi.storage.postgres.manager import pg_manager
+        from yuxi.storage.postgres.models_business import AgentRun
+
+        uid = _get_runtime_uid(runtime)
+        await _check_session_access(uid, thread_id)
+
+        async with pg_manager.get_async_session_context() as db:
+            result = await db.execute(
+                select(AgentRun)
+                .where(
+                    AgentRun.thread_id == thread_id,
+                    AgentRun.uid == uid,
+                )
+                .order_by(AgentRun.created_at.desc())
+                .limit(1)
+            )
+            run = result.scalar_one_or_none()
+
+        if not run:
+            return f"会话 {thread_id} 暂无运行记录"
+
+        lines = [
+            f"运行 ID: {run.id}",
+            f"状态: {run.status}",
+            f"类型: {run.run_type}",
+            f"智能体: {run.agent_id}",
+            f"创建时间: {run.created_at}",
+        ]
+        if run.started_at:
+            lines.append(f"开始时间: {run.started_at}")
+        if run.finished_at:
+            lines.append(f"完成时间: {run.finished_at}")
+        if run.error_message:
+            lines.append(f"错误: {run.error_message}")
+
+        return "\n".join(lines)
diff --git a/backend/package/yuxi/repositories/conversation_repository.py b/backend/package/yuxi/repositories/conversation_repository.py
index 4b3f77f6..9aabb9d7 100644
--- a/backend/package/yuxi/repositories/conversation_repository.py
+++ b/backend/package/yuxi/repositories/conversation_repository.py
@@ -73,6 +73,20 @@ class ConversationRepository:
         result = await self.db.execute(select(Conversation).where(Conversation.thread_id == thread_id))
         return result.scalar_one_or_none()
 
+    async def get_conversations_by_parent_thread_id(
+        self, parent_thread_id: str, limit: int | None = None
+    ) -> list[Conversation]:
+        """查询指定父会话下的子会话列表。"""
+        query = (
+            select(Conversation)
+            .where(Conversation.parent_thread_id == parent_thread_id)
+            .order_by(Conversation.created_at.desc())
+        )
+        if limit is not None:
+            query = query.limit(limit)
+        result = await self.db.execute(query)
+        return list(result.scalars().all())
+
     async def _get_conversation_by_id(self, conversation_id: int) -> Conversation | None:
         result = await self.db.execute(select(Conversation).where(Conversation.id == conversation_id))
         return result.scalar_one_or_none()
diff --git a/backend/package/yuxi/storage/postgres/manager.py b/backend/package/yuxi/storage/postgres/manager.py
index 09414a64..a875eef6 100644
--- a/backend/package/yuxi/storage/postgres/manager.py
+++ b/backend/package/yuxi/storage/postgres/manager.py
@@ -465,6 +465,9 @@ class PostgresManager(metaclass=SingletonMeta):
             "CREATE INDEX IF NOT EXISTS ix_conversations_is_pinned ON conversations(is_pinned)",
             "CREATE UNIQUE INDEX IF NOT EXISTS ix_model_providers_provider_id ON model_providers(provider_id)",
             "CREATE INDEX IF NOT EXISTS ix_model_providers_is_enabled ON model_providers(is_enabled)",
+            "ALTER TABLE IF EXISTS conversations ADD COLUMN IF NOT EXISTS parent_thread_id VARCHAR(64)",
+            "ALTER TABLE IF EXISTS conversations ADD COLUMN IF NOT EXISTS spawn_depth INTEGER NOT NULL DEFAULT 0",
+            "CREATE INDEX IF NOT EXISTS idx_conversations_parent ON conversations(parent_thread_id)",
         ]
         async with self.async_engine.begin() as conn:
             for stmt in stmts:
diff --git a/backend/package/yuxi/storage/postgres/models_business.py b/backend/package/yuxi/storage/postgres/models_business.py
index 8b18401a..67e04b18 100644
--- a/backend/package/yuxi/storage/postgres/models_business.py
+++ b/backend/package/yuxi/storage/postgres/models_business.py
@@ -263,6 +263,8 @@ class Conversation(Base):
     created_at = Column(DateTime, default=utc_now_naive, comment="Creation time")
     updated_at = Column(DateTime, default=utc_now_naive, onupdate=utc_now_naive, comment="Update time")
     extra_metadata = Column(JSON, nullable=True, comment="Additional metadata")
+    parent_thread_id = Column(String(64), nullable=True, index=True, comment="Parent thread ID for sub-agent sessions")
+    spawn_depth = Column(Integer, default=0, nullable=False, comment="Nesting depth: 0=main, 1=first-level sub-agent, etc.")
 
     # Relationships
     messages = relationship("Message", back_populates="conversation", cascade="all, delete-orphan")
@@ -283,6 +285,8 @@ class Conversation(Base):
             "created_at": format_utc_datetime(self.created_at),
             "updated_at": format_utc_datetime(self.updated_at),
             "metadata": metadata,
+            "parent_thread_id": self.parent_thread_id,
+            "spawn_depth": self.spawn_depth,
         }
 
 
diff --git a/backend/test/unit/agents/toolkits/buildin/test_session_tools.py b/backend/test/unit/agents/toolkits/buildin/test_session_tools.py
new file mode 100644
index 00000000..2990d9b1
--- /dev/null
+++ b/backend/test/unit/agents/toolkits/buildin/test_session_tools.py
@@ -0,0 +1,470 @@
+"""会话通信工具集单元测试"""
+
+import os
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from yuxi.agents.toolkits.buildin import session_tools
+
+
+# ── 辅助函数测试 ──────────────────────────────────────────
+
+
+def _make_runtime(uid: str = "user-1", thread_id: str = "thread-1", file_thread_id: str | None = None):
+    """构造模拟的 ToolRuntime 对象。"""
+    context = MagicMock()
+    context.uid = uid
+    context.thread_id = thread_id
+    context.file_thread_id = file_thread_id or thread_id
+    runtime = MagicMock()
+    runtime.context = context
+    return runtime
+
+
+class TestGetRuntimeUid:
+    def test_returns_uid(self):
+        runtime = _make_runtime(uid="user-123")
+        assert session_tools._get_runtime_uid(runtime) == "user-123"
+
+    def test_raises_when_uid_missing(self):
+        runtime = _make_runtime(uid="")
+        with pytest.raises(ValueError, match="uid"):
+            session_tools._get_runtime_uid(runtime)
+
+    def test_raises_when_uid_none(self):
+        runtime = _make_runtime(uid=None)
+        with pytest.raises(ValueError, match="uid"):
+            session_tools._get_runtime_uid(runtime)
+
+
+class TestGetRuntimeThreadId:
+    def test_returns_thread_id(self):
+        runtime = _make_runtime(thread_id="thread-1")
+        assert session_tools._get_runtime_thread_id(runtime) == "thread-1"
+
+    def test_ignores_file_thread_id(self):
+        """file_thread_id 继承自父会话，不应作为当前会话标识"""
+        runtime = _make_runtime(thread_id="thread-1", file_thread_id="file-thread-1")
+        assert session_tools._get_runtime_thread_id(runtime) == "thread-1"
+
+    def test_raises_when_missing(self):
+        runtime = _make_runtime(thread_id="", file_thread_id=None)
+        with pytest.raises(ValueError, match="thread_id"):
+            session_tools._get_runtime_thread_id(runtime)
+
+
+# ── 权限校验测试 ──────────────────────────────────────────
+
+
+class TestCheckSessionAccess:
+    @pytest.mark.asyncio
+    async def test_allows_own_conversation(self):
+        """用户访问自己的会话 → 允许"""
+        conv = MagicMock()
+        conv.uid = "user-1"
+        conv.parent_thread_id = None
+
+        with patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.get_conversation_by_thread_id = AsyncMock(return_value=conv)
+            MockPg.get_async_session_context = MagicMock()
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            # Should not raise
+            await session_tools._check_session_access("user-1", "thread-1")
+
+    @pytest.mark.asyncio
+    async def test_allows_via_parent_chain(self):
+        """沿 parent 链上溯找到自己的会话 → 允许"""
+        child_conv = MagicMock()
+        child_conv.uid = "other-user"
+        child_conv.parent_thread_id = "parent-thread"
+
+        parent_conv = MagicMock()
+        parent_conv.uid = "user-1"
+        parent_conv.parent_thread_id = None
+
+        with patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.get_conversation_by_thread_id = AsyncMock(side_effect=[child_conv, parent_conv])
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            await session_tools._check_session_access("user-1", "child-thread")
+
+    @pytest.mark.asyncio
+    async def test_rejects_unauthorized_conversation(self):
+        """非自己的会话且不在子会话树内 → 拒绝"""
+        conv = MagicMock()
+        conv.uid = "other-user"
+        conv.parent_thread_id = None
+
+        with patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.get_conversation_by_thread_id = AsyncMock(return_value=conv)
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            with pytest.raises(PermissionError, match="无权访问"):
+                await session_tools._check_session_access("user-1", "other-thread")
+
+    @pytest.mark.asyncio
+    async def test_rejects_nonexistent_conversation(self):
+        """会话不存在 → 抛出 ValueError"""
+        with patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.get_conversation_by_thread_id = AsyncMock(return_value=None)
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            with pytest.raises(ValueError, match="会话不存在"):
+                await session_tools._check_session_access("user-1", "nonexistent")
+
+    @pytest.mark.asyncio
+    async def test_prevents_infinite_loop_in_parent_chain(self):
+        """parent 链出现循环时不会无限循环"""
+        conv_a = MagicMock()
+        conv_a.uid = "other-user"
+        conv_a.parent_thread_id = "thread-b"
+
+        conv_b = MagicMock()
+        conv_b.uid = "other-user"
+        conv_b.parent_thread_id = "thread-a"  # 循环回 A
+
+        with patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.get_conversation_by_thread_id = AsyncMock(side_effect=[conv_a, conv_b])
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            with pytest.raises(PermissionError, match="无权访问"):
+                await session_tools._check_session_access("user-1", "thread-a")
+
+
+# ── list_sessions 测试 ────────────────────────────────────
+
+
+class TestListSessions:
+    @pytest.mark.asyncio
+    async def test_returns_formatted_list(self):
+        conv = MagicMock()
+        conv.thread_id = "thread-1"
+        conv.agent_id = "agent-1"
+        conv.status = "active"
+        conv.title = "测试会话"
+        conv.parent_thread_id = None
+        conv.updated_at = "2026-06-13"
+
+        with patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_conversations = AsyncMock(return_value=[conv])
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            result = await session_tools.list_sessions.coroutine(
+                runtime=_make_runtime(),
+            )
+            assert "thread-1" in result
+            assert "agent-1" in result
+            assert "测试会话" in result
+            assert "[子智能体]" not in result
+
+    @pytest.mark.asyncio
+    async def test_marks_subagent_sessions(self):
+        conv = MagicMock()
+        conv.thread_id = "sub-thread-1"
+        conv.agent_id = "sub-agent"
+        conv.status = "active"
+        conv.title = "子智能体会话"
+        conv.parent_thread_id = "parent-thread"
+        conv.updated_at = "2026-06-13"
+
+        with patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_conversations = AsyncMock(return_value=[conv])
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            result = await session_tools.list_sessions.coroutine(
+                runtime=_make_runtime(),
+            )
+            assert "[子智能体]" in result
+
+    @pytest.mark.asyncio
+    async def test_returns_empty_message(self):
+        with patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_conversations = AsyncMock(return_value=[])
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            result = await session_tools.list_sessions.coroutine(
+                runtime=_make_runtime(),
+            )
+            assert "无可见" in result
+
+
+# ── get_session_history 测试 ──────────────────────────────
+
+
+class TestGetSessionHistory:
+    @pytest.mark.asyncio
+    async def test_returns_messages(self):
+        msg = MagicMock()
+        msg.role = "user"
+        msg.content = "你好"
+        msg.message_type = "text"
+
+        conv = MagicMock()
+        conv.id = 1
+
+        mock_db = MagicMock()
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.unique.return_value.all.return_value = [msg]
+        mock_db.execute = AsyncMock(return_value=mock_result)
+
+        with patch("yuxi.agents.toolkits.buildin.session_tools._check_session_access", new_callable=AsyncMock), \
+             patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.get_conversation_by_thread_id = AsyncMock(return_value=conv)
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=mock_db)
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            result = await session_tools.get_session_history.coroutine(
+                thread_id="thread-1",
+                runtime=_make_runtime(),
+            )
+            assert "[user] 你好" in result
+
+    @pytest.mark.asyncio
+    async def test_filters_tool_messages_by_default(self):
+        msg_tool = MagicMock()
+        msg_tool.role = "assistant"
+        msg_tool.content = "tool result"
+        msg_tool.message_type = "tool_result"
+
+        msg_text = MagicMock()
+        msg_text.role = "user"
+        msg_text.content = "hello"
+        msg_text.message_type = "text"
+
+        conv = MagicMock()
+        conv.id = 1
+
+        mock_db = MagicMock()
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.unique.return_value.all.return_value = [msg_tool, msg_text]
+        mock_db.execute = AsyncMock(return_value=mock_result)
+
+        with patch("yuxi.agents.toolkits.buildin.session_tools._check_session_access", new_callable=AsyncMock), \
+             patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.get_conversation_by_thread_id = AsyncMock(return_value=conv)
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=mock_db)
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            result = await session_tools.get_session_history.coroutine(
+                thread_id="thread-1",
+                runtime=_make_runtime(),
+            )
+            assert "tool result" not in result
+            assert "hello" in result
+
+    @pytest.mark.asyncio
+    async def test_includes_tool_messages_when_requested(self):
+        msg_tool = MagicMock()
+        msg_tool.role = "assistant"
+        msg_tool.content = "tool result"
+        msg_tool.message_type = "tool_result"
+
+        conv = MagicMock()
+        conv.id = 1
+
+        mock_db = MagicMock()
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.unique.return_value.all.return_value = [msg_tool]
+        mock_db.execute = AsyncMock(return_value=mock_result)
+
+        with patch("yuxi.agents.toolkits.buildin.session_tools._check_session_access", new_callable=AsyncMock), \
+             patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.get_conversation_by_thread_id = AsyncMock(return_value=conv)
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=mock_db)
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            result = await session_tools.get_session_history.coroutine(
+                thread_id="thread-1",
+                include_tools=True,
+                runtime=_make_runtime(),
+            )
+            assert "tool result" in result
+
+    @pytest.mark.asyncio
+    async def test_truncates_long_content(self):
+        msg = MagicMock()
+        msg.role = "assistant"
+        msg.content = "x" * 5000
+        msg.message_type = "text"
+
+        conv = MagicMock()
+        conv.id = 1
+
+        mock_db = MagicMock()
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.unique.return_value.all.return_value = [msg]
+        mock_db.execute = AsyncMock(return_value=mock_result)
+
+        with patch("yuxi.agents.toolkits.buildin.session_tools._check_session_access", new_callable=AsyncMock), \
+             patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.get_conversation_by_thread_id = AsyncMock(return_value=conv)
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=mock_db)
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            result = await session_tools.get_session_history.coroutine(
+                thread_id="thread-1",
+                runtime=_make_runtime(),
+            )
+            assert "已截断" in result
+
+    @pytest.mark.asyncio
+    async def test_returns_empty_message_when_no_conversation(self):
+        with patch("yuxi.agents.toolkits.buildin.session_tools._check_session_access", new_callable=AsyncMock), \
+             patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.get_conversation_by_thread_id = AsyncMock(return_value=None)
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            result = await session_tools.get_session_history.coroutine(
+                thread_id="thread-1",
+                runtime=_make_runtime(),
+            )
+            assert "暂无消息" in result
+
+
+# ── send_to_session 测试 ──────────────────────────────────
+
+
+class TestSendToSession:
+    @pytest.mark.asyncio
+    async def test_sends_message_with_annotation(self):
+        with patch("yuxi.agents.toolkits.buildin.session_tools._check_session_access", new_callable=AsyncMock), \
+             patch("yuxi.agents.toolkits.buildin.session_tools.ConversationRepository") as MockRepo, \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_repo = MockRepo.return_value
+            mock_repo.add_message_by_thread_id = AsyncMock(return_value=MagicMock())
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            result = await session_tools.send_to_session.coroutine(
+                thread_id="other-thread",
+                message="hello",
+                runtime=_make_runtime(thread_id="my-thread"),
+            )
+            assert "已发送" in result
+            # 验证消息带有来源标注
+            call_args = mock_repo.add_message_by_thread_id.call_args
+            assert "[来自会话 my-thread]" in call_args.kwargs["content"]
+            assert call_args.kwargs["extra_metadata"]["source_type"] == "inter_session"
+            assert call_args.kwargs["extra_metadata"]["source_thread_id"] == "my-thread"
+
+    @pytest.mark.asyncio
+    async def test_rejects_sending_to_self(self):
+        """禁止向自身会话发送消息"""
+        result = await session_tools.send_to_session.coroutine(
+            thread_id="my-thread",
+            message="hello",
+            runtime=_make_runtime(thread_id="my-thread"),
+        )
+        assert "不能向当前所在的会话发送消息" in result
+
+
+# ── get_agent_progress 测试 ───────────────────────────────
+
+
+class TestGetAgentProgress:
+    @pytest.mark.asyncio
+    async def test_returns_run_info(self):
+        run = MagicMock()
+        run.id = "run-123"
+        run.status = "running"
+        run.run_type = "chat"
+        run.agent_id = "agent-1"
+        run.created_at = "2026-06-13"
+        run.started_at = "2026-06-13T10:00"
+        run.finished_at = None
+        run.error_message = None
+
+        with patch("yuxi.agents.toolkits.buildin.session_tools._check_session_access", new_callable=AsyncMock), \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_db = MagicMock()
+            mock_result = MagicMock()
+            mock_result.scalar_one_or_none.return_value = run
+            mock_db.execute = AsyncMock(return_value=mock_result)
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=mock_db)
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            result = await session_tools.get_agent_progress.coroutine(
+                thread_id="thread-1",
+                runtime=_make_runtime(),
+            )
+            assert "run-123" in result
+            assert "running" in result
+
+    @pytest.mark.asyncio
+    async def test_returns_no_run_message(self):
+        with patch("yuxi.agents.toolkits.buildin.session_tools._check_session_access", new_callable=AsyncMock), \
+             patch("yuxi.agents.toolkits.buildin.session_tools.pg_manager") as MockPg:
+            mock_db = MagicMock()
+            mock_result = MagicMock()
+            mock_result.scalar_one_or_none.return_value = None
+            mock_db.execute = AsyncMock(return_value=mock_result)
+            MockPg.get_async_session_context.return_value.__aenter__ = AsyncMock(return_value=mock_db)
+            MockPg.get_async_session_context.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            result = await session_tools.get_agent_progress.coroutine(
+                thread_id="thread-1",
+                runtime=_make_runtime(),
+            )
+            assert "暂无运行记录" in result
+
+
+# ── LITE 模式测试 ─────────────────────────────────────────
+
+
+class TestLiteMode:
+    def test_lite_mode_flag(self):
+        """验证 LITE_MODE 常量正确读取环境变量"""
+        # 当前测试环境未设置 LITE_MODE，应为 False
+        assert session_tools._LITE_MODE is False
+
+    def test_lite_mode_tools_not_in_global_registry(self):
+        """非 LITE 模式下，工具应已注册到全局列表"""
+        from yuxi.agents.toolkits.registry import _all_tool_instances, _extra_registry
+
+        tool_names = {t.name for t in _all_tool_instances}
+        assert "list_sessions" in tool_names
+        assert "get_session_history" in tool_names
+        assert "send_to_session" in tool_names
+        assert "get_agent_progress" in tool_names
+
+        assert "list_sessions" in _extra_registry
+        assert "get_session_history" in _extra_registry
+        assert "send_to_session" in _extra_registry
+        assert "get_agent_progress" in _extra_registry