feat: /agent command for per-group/per-user AI agent switching (#129)

Add /agent slash command that allows admins to switch the AI agent used
in a specific group or personal chat. Each group/user stores their own
preference independently via active_agent_id FK on bot_users/bot_groups.

Also fix MCP tool permission issue in bypassPermissions mode where
ctos_user_id was not injected into tool parameters, causing "permission
denied" errors for search_nas_files and read_document.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: yazelin

backend/migrations/versions/011_add_active_agent_id.py

@@ -0,0 +1,60 @@
+"""新增 bot_users 和 bot_groups 的 active_agent_id 欄位
+
+用於 /agent 指令切換對話使用的 AI Agent。
+
+Revision ID: 011
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "011"
+down_revision = "010"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # bot_users 新增 active_agent_id
+    op.add_column(
+        "bot_users",
+        sa.Column(
+            "active_agent_id",
+            sa.dialects.postgresql.UUID(as_uuid=True),
+            nullable=True,
+        ),
+    )
+    op.create_foreign_key(
+        "fk_bot_users_active_agent_id",
+        "bot_users",
+        "ai_agents",
+        ["active_agent_id"],
+        ["id"],
+        ondelete="SET NULL",
+    )
+
+    # bot_groups 新增 active_agent_id
+    op.add_column(
+        "bot_groups",
+        sa.Column(
+            "active_agent_id",
+            sa.dialects.postgresql.UUID(as_uuid=True),
+            nullable=True,
+        ),
+    )
+    op.create_foreign_key(
+        "fk_bot_groups_active_agent_id",
+        "bot_groups",
+        "ai_agents",
+        ["active_agent_id"],
+        ["id"],
+        ondelete="SET NULL",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint("fk_bot_groups_active_agent_id", "bot_groups", type_="foreignkey")
+    op.drop_column("bot_groups", "active_agent_id")
+    op.drop_constraint("fk_bot_users_active_agent_id", "bot_users", type_="foreignkey")
+    op.drop_column("bot_users", "active_agent_id")

backend/src/ching_tech_os/services/ai_manager.py

@@ -350,6 +350,24 @@ async def get_agent_by_name(name: str) -> dict | None:
         return result
 
 
+async def get_selectable_agents() -> list[dict]:
+    """取得可供 /agent 指令切換的 Agent 清單
+
+    查詢 is_active=true 且 settings.user_selectable='true' 的 Agent，按 name 排序。
+    """
+    async with get_connection() as conn:
+        rows = await conn.fetch(
+            """
+            SELECT id, name, display_name, description, model
+            FROM ai_agents
+            WHERE is_active = true
+              AND settings->>'user_selectable' = 'true'
+            ORDER BY name
+            """,
+        )
+        return [dict(row) for row in rows]
+
+
 async def create_agent(data: AiAgentCreate) -> dict:
     """建立 Agent"""
     settings_json = json.dumps(data.settings) if data.settings else None

backend/src/ching_tech_os/services/bot/command_handlers.py

@@ -193,6 +193,95 @@ async def _handle_debug(ctx: CommandContext) -> str | None:
     return reply_text or "診斷完成，但未產生回報內容。"
 
 
+async def _handle_agent(ctx: CommandContext) -> str | None:
+    """切換對話使用的 AI Agent
+
+    用法：
+    /agent          — 顯示目前使用的 Agent 和可切換清單
+    /agent <name>   — 用名稱切換
+    /agent <number> — 用編號切換
+    /agent reset    — 恢復預設
+    """
+    from .. import ai_manager
+    from ..linebot_agents import (
+        get_group_active_agent_id,
+        get_user_active_agent_id,
+        set_group_active_agent,
+        set_user_active_agent,
+    )
+
+    args = ctx.raw_args.strip()
+
+    # 取得可切換的 Agent 清單（按 name 排序）
+    selectable = await ai_manager.get_selectable_agents()
+
+    async def _apply_preference(agent_id: str | None) -> None:
+        """將 Agent 偏好寫入群組或個人"""
+        if ctx.is_group and ctx.group_id:
+            await set_group_active_agent(ctx.group_id, agent_id)
+        elif ctx.bot_user_id:
+            await set_user_active_agent(ctx.bot_user_id, agent_id)
+
+    # === /agent reset ===
+    if args.lower() == "reset":
+        await _apply_preference(None)
+        return "已恢復預設 Agent"
+
+    # === 查詢目前 Agent ===
+    current_agent_id = None
+    if ctx.is_group and ctx.group_id:
+        current_agent_id = await get_group_active_agent_id(ctx.group_id)
+    elif ctx.bot_user_id:
+        current_agent_id = await get_user_active_agent_id(ctx.bot_user_id)
+
+    # 取得目前 Agent 的顯示資訊
+    current_label = "預設"
+    if current_agent_id:
+        from uuid import UUID
+        current_agent = await ai_manager.get_agent(UUID(current_agent_id))
+        if current_agent:
+            current_label = f"{current_agent['name']}（{current_agent.get('display_name', '')}）"
+        else:
+            current_label = "預設（偏好 Agent 已不存在）"
+
+    # === /agent（無參數）— 顯示狀態和清單 ===
+    if not args:
+        lines = [f"目前 Agent：{current_label}"]
+        if selectable:
+            lines.append("")
+            lines.append("可切換的 Agent：")
+            for i, agent in enumerate(selectable, 1):
+                display = agent.get("display_name") or agent["name"]
+                lines.append(f"{i}. {agent['name']} — {display}")
+            lines.append("")
+            lines.append("用法：/agent <名稱或編號>")
+            lines.append("恢復預設：/agent reset")
+        else:
+            lines.append("目前沒有可切換的 Agent")
+            lines.append("請在 AI 管理介面將 Agent 的 settings.user_selectable 設為 true")
+        return "\n".join(lines)
+
+    # === /agent <number> — 編號切換 ===
+    if args.isdigit():
+        idx = int(args)
+        if idx < 1 or idx > len(selectable):
+            return f"編號 {idx} 超出範圍（1-{len(selectable)}），請用 /agent 查看可用清單"
+        target = selectable[idx - 1]
+    else:
+        # === /agent <name> — 名稱切換 ===
+        target = next((a for a in selectable if a["name"] == args), None)
+        if not target:
+            # 檢查 Agent 是否存在但不可選
+            existing = await ai_manager.get_agent_by_name(args)
+            if existing:
+                return f"Agent {args} 不可切換，請用 /agent 查看可用清單"
+            return f"找不到 Agent: {args}，請用 /agent 查看可用清單"
+
+    await _apply_preference(str(target["id"]))
+    display = target.get("display_name") or target["name"]
+    return f"已切換到 {display}"
+
+
 _registered = False
 
 
@@ -243,6 +332,15 @@ def register_builtin_commands() -> None:
             require_admin=True,
             private_only=True,
         ),
+        SlashCommand(
+            name="agent",
+            aliases=["切換助理"],
+            handler=_handle_agent,
+            description="切換 AI Agent",
+            require_bound=True,
+            require_admin=True,
+            private_only=False,
+        ),
     ]
 
     for cmd in all_commands:

backend/src/ching_tech_os/services/bot_telegram/handler.py

@@ -680,7 +680,11 @@ async def _handle_text_with_ai(
         logger.error(f"取得對話歷史失敗: {e}", exc_info=True)
 
     # 1. 取得 Agent 設定
-    agent = await get_linebot_agent(is_group=is_group)
+    agent = await get_linebot_agent(
+        is_group=is_group,
+        bot_user_id=bot_user_id,
+        bot_group_id=bot_group_id,
+    )
     if not agent:
         logger.error("找不到 Agent 設定")
         await adapter.send_text(chat_id, "系統尚未設定 AI Agent，請聯繫管理員。")

backend/src/ching_tech_os/services/linebot_agents.py

@@ -635,16 +635,96 @@ async def ensure_default_linebot_agents() -> None:
     await _ensure_agents(DEFAULT_BOT_MODE_AGENTS, use_dynamic_prompt=False)
 
 
-async def get_linebot_agent(is_group: bool) -> dict | None:
-    """
-    取得 Line Bot Agent 設定。
+async def set_user_active_agent(bot_user_id: str, agent_id: str | None) -> None:
+    """設定用戶的個人對話 Agent 偏好"""
+    from uuid import UUID as _UUID
+
+    from ..database import get_connection
+
+    value = _UUID(agent_id) if agent_id else None
+    async with get_connection() as conn:
+        await conn.execute(
+            "UPDATE bot_users SET active_agent_id = $1 WHERE id = $2",
+            value,
+            bot_user_id,
+        )
+
+
+async def set_group_active_agent(bot_group_id: str, agent_id: str | None) -> None:
+    """設定群組的 Agent 偏好"""
+    from uuid import UUID as _UUID
+
+    from ..database import get_connection
+
+    value = _UUID(agent_id) if agent_id else None
+    async with get_connection() as conn:
+        await conn.execute(
+            "UPDATE bot_groups SET active_agent_id = $1 WHERE id = $2",
+            value,
+            bot_group_id,
+        )
+
+
+async def get_user_active_agent_id(bot_user_id: str) -> str | None:
+    """查詢用戶的 active_agent_id"""
+    from ..database import get_connection
+
+    async with get_connection() as conn:
+        row = await conn.fetchrow(
+            "SELECT active_agent_id FROM bot_users WHERE id = $1",
+            bot_user_id,
+        )
+        return str(row["active_agent_id"]) if row and row["active_agent_id"] else None
+
+
+async def get_group_active_agent_id(bot_group_id: str) -> str | None:
+    """查詢群組的 active_agent_id"""
+    from ..database import get_connection
+
+    async with get_connection() as conn:
+        row = await conn.fetchrow(
+            "SELECT active_agent_id FROM bot_groups WHERE id = $1",
+            bot_group_id,
+        )
+        return str(row["active_agent_id"]) if row and row["active_agent_id"] else None
+
+
+async def get_linebot_agent(
+    is_group: bool,
+    *,
+    bot_user_id: str | None = None,
+    bot_group_id: str | None = None,
+) -> dict | None:
+    """取得 Line Bot Agent 設定，支援偏好覆蓋。
+
+    路由優先級：
+    1. 群組對話：bot_groups.active_agent_id > 預設 linebot-group
+    2. 個人對話：bot_users.active_agent_id > 預設 linebot-personal
 
     Args:
         is_group: 是否為群組對話
+        bot_user_id: Bot 用戶 ID（用於查詢個人偏好）
+        bot_group_id: Bot 群組 ID（用於查詢群組偏好）
 
     Returns:
         Agent 設定字典，包含 model 和 system_prompt
         如果找不到則回傳 None
     """
+    from uuid import UUID
+
+    # 查詢偏好 Agent
+    active_agent_id = None
+    if is_group and bot_group_id:
+        active_agent_id = await get_group_active_agent_id(bot_group_id)
+    elif not is_group and bot_user_id:
+        active_agent_id = await get_user_active_agent_id(bot_user_id)
+
+    if active_agent_id:
+        agent = await ai_manager.get_agent(UUID(active_agent_id))
+        if agent and agent.get("is_active"):
+            return agent
+        # 偏好 Agent 不存在或已停用，fallback 到預設
+        logger.warning(f"偏好 Agent {active_agent_id} 不可用，使用預設 Agent")
+
     agent_name = AGENT_LINEBOT_GROUP if is_group else AGENT_LINEBOT_PERSONAL
     return await ai_manager.get_agent_by_name(agent_name)

backend/src/ching_tech_os/services/linebot_ai.py

@@ -600,6 +600,7 @@ async def process_message_with_ai(
     reply_token: str | None,
     user_display_name: str | None = None,
     quoted_message_id: str | None = None,
+    bot_user_id: str | None = None,
 ) -> str | None:
     """
     使用 AI 處理訊息
@@ -637,16 +638,25 @@ async def process_message_with_ai(
 
     try:
         # 取得 Agent 設定
-        agent = await get_linebot_agent(is_group)
-        agent_name = AGENT_LINEBOT_GROUP if is_group else AGENT_LINEBOT_PERSONAL
+        # 群組 ID 轉換為字串（bot_groups.id 是 UUID）
+        bot_group_id_str = str(line_group_id) if line_group_id else None
+        agent = await get_linebot_agent(
+            is_group,
+            bot_user_id=bot_user_id,
+            bot_group_id=bot_group_id_str,
+        )
 
         if not agent:
-            error_msg = f"⚠️ AI 設定錯誤：Agent '{agent_name}' 不存在"
+            fallback_name = AGENT_LINEBOT_GROUP if is_group else AGENT_LINEBOT_PERSONAL
+            error_msg = f"⚠️ AI 設定錯誤：Agent '{fallback_name}' 不存在"
             logger.error(error_msg)
             if reply_token:
                 await reply_text(reply_token, error_msg)
             return error_msg
 
+        # agent 保證非 None
+        agent_name = agent.get("name", "")
+
         # 從 Agent 取得 model 和基礎 prompt
         model = agent.get("model", "opus").replace("claude-", "")  # claude-sonnet -> sonnet
         # 安全取得 system_prompt（處理 None 和非 dict 情況）
@@ -1590,4 +1600,5 @@ async def handle_text_message(
         reply_token=reply_token,
         user_display_name=user_display_name,
         quoted_message_id=quoted_message_id,
+        bot_user_id=bot_user_id,
     )

backend/src/ching_tech_os/services/mcp/nas_tools.py

@@ -28,6 +28,10 @@
 async def _get_user_shared_mounts(ctos_user_id: int | None) -> dict[str, str]:
     """取得使用者可存取的 shared 掛載點。"""
     from ..path_manager import path_manager
+    from .server import resolve_ctos_user_id
+
+    # bypassPermissions 模式下 AI 可能不傳 ctos_user_id，fallback 環境變數
+    ctos_user_id = resolve_ctos_user_id(ctos_user_id)
 
     return await get_allowed_shared_mounts_for_user(
         path_manager.get_shared_mounts(),