feat(agent-gate): 智能体互通编排层——authorize_call 三层权限(L1管理员/L2同企业/L3临时授权)+MQTT invoke 派发+agent-events 内部接口+agent-grants CRUD+幂等迁移SQL

app/agent_gate/__init__.py

@@ -0,0 +1,207 @@
+# -*- coding: utf-8 -*-
+"""智能体互通编排（server-core 权威）。
+
+三层权限模型（全部在服务端集中校验，客户端不可绕过）：
+
+- **L1 企业管理员无条件调用**：企业管理员（company_members.is_admin/member_type∈{admin,owner}）
+  可调用**本企业任意成员**的智能体，无需授权记录。
+- **L2 企业内部智能体互调**：同一企业内的账号可互相调用彼此的智能体（企业=信任域）。
+- **L3 成员开放调用（临时授权）**：成员 A 通过 agent_access_grants 临时授权成员 B 调用
+  A 的智能体（限时 / 限次 / 限 prompt 前缀 scope）。
+
+通用约束：目标 agent 的 ``external_callable=False`` 时拒绝一切外部调用（agent 归属用户
+可完全关闭外部调用）；``call_mode=confirm`` 时由客户端在执行前向归属用户请求确认。
+"""
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+
+from ..infrastructure.repositories import Database
+
+log = logging.getLogger("agent_gate")
+
+AGENT_TOPIC_PREFIX = "opc/agent/v1"
+
+# 归属用户本人调用自己的智能体，永远允许
+SELF_ALLOWED = True
+
+
+def parse_agent_addr(addr: str) -> tuple[str, str] | None:
+    """解析 ``{user_id}.{agent_id}`` → (user_id, agent_id)；非法返回 None。"""
+    if not addr or "." not in addr:
+        return None
+    user_id, agent_id = addr.split(".", 1)
+    if not user_id or not agent_id:
+        return None
+    return user_id, agent_id
+
+
+def _is_expired(expires_at: str) -> bool:
+    if not expires_at:
+        return False
+    try:
+        exp = datetime.fromisoformat(str(expires_at).replace("Z", "+00:00"))
+        if exp.tzinfo is None:
+            exp = exp.replace(tzinfo=timezone.utc)
+        return exp < datetime.now(timezone.utc)
+    except Exception:  # noqa: BLE001
+        return True  # 无法解析视为已过期（保守）
+
+def _scope_allows(scope: str, prompt: str) -> bool:
+    """scope 为 prompt 前缀白名单（每行一条前缀）；空=全部允许。"""
+    if not scope:
+        return True
+    for line in scope.splitlines():
+        prefix = line.strip()
+        if prefix and prompt.startswith(prefix):
+            return True
+    return False
+
+
+async def authorize_call(db: Database, caller_user_id: str, target_addr: str,
+                         prompt: str = "") -> tuple[bool, str]:
+    """校验调用权限。返回 (ok, reason)。
+
+    优先级：自己 → L1 管理员 → L2 同企业 → L3 显式授权 → 拒绝。
+    """
+    parsed = parse_agent_addr(target_addr)
+    if parsed is None:
+        return False, "目标智能体地址无效"
+    target_user_id, agent_id = parsed
+
+    # 自己调自己的智能体
+    if SELF_ALLOWED and caller_user_id == target_user_id:
+        return True, ""
+
+    # 目标 agent 是否允许外部调用
+    agent = await db.agents.get(agent_id, target_user_id, port="opc")
+    if agent is None:
+        # 未注册到平台（可能未登录过桌面端），视为不可调用
+        return False, "目标智能体不存在或未启用"
+    if not agent.get("external_callable", True):
+        return False, "该智能体已关闭外部调用"
+
+    caller_companies = await db.company_members.list_by_user(caller_user_id)
+    if not caller_companies:
+        return False, "无调用权限（非企业成员）"
+
+    target_companies = await db.company_members.list_by_user(target_user_id)
+    target_company_ids = {c["company_id"] for c in target_companies if c.get("status", "active") == "active"}
+
+    for cm in caller_companies:
+        if cm.get("status", "active") != "active":
+            continue
+        company_id = cm["company_id"]
+        # L1：企业管理员 → 本企业成员 agent
+        is_admin = bool(cm.get("is_admin")) or cm.get("member_type") in ("admin", "owner")
+        if is_admin and company_id in target_company_ids:
+            return True, ""
+        # L2：同一企业互调
+        if company_id in target_company_ids:
+            return True, ""
+
+    # L3：显式授权
+    grant = await db.agent_access_grants.get_active(
+        grantor_user_id=target_user_id,
+        target_agent_addr=target_addr,
+        grantee_user_id=caller_user_id,
+    )
+    if grant is None:
+        return False, "无调用权限（需要企业管理员或临时授权）"
+    if _is_expired(grant.expires_at):
+        return False, "授权已过期"
+    if grant.max_calls > 0 and (grant.used_calls or 0) >= grant.max_calls:
+        return False, "授权调用次数已用完"
+    if not _scope_allows(grant.scope, prompt):
+        return False, "调用内容超出授权范围"
+
+    # 计数（派发即计数）
+    await db.agent_access_grants.consume_call(grant)
+    return True, ""
+
+
+async def handle_agent_task_event(
+    db: Database,
+    im,
+    *,
+    conv_id: str,
+    sender_id: str,
+    sender_name: str,
+    content: str,
+    metadata: dict | None,
+) -> dict:
+    """处理 im-service 的 agent_task 消息事件（管理员/成员在 IM 里发起调用）。
+
+    流程：解析地址 → 权限校验 → 目标 agent 存在性 → 在线查询 →
+    在线则派发 invoke（im 发布 MQTT），离线则回写 offline 回执。
+    """
+    metadata = metadata or {}
+    target_addr = str(metadata.get("target_agent_addr") or metadata.get("target_addr") or "").strip()
+    if not target_addr:
+        return {"ok": False, "reason": "缺少目标智能体地址（target_agent_addr）"}
+
+    parsed = parse_agent_addr(target_addr)
+    if parsed is None:
+        return {"ok": False, "reason": "目标智能体地址无效"}
+    target_user_id, agent_id = parsed
+
+    prompt = str(content or metadata.get("prompt") or "").strip()
+    if not prompt:
+        return {"ok": False, "reason": "缺少调用内容"}
+
+    ok, reason = await authorize_call(db, sender_id, target_addr, prompt)
+    if not ok:
+        # 无权限：向会话回写一条 agent_reply 状态消息
+        try:
+            await im.internal_agent_reply(conv_id, "", "error", f"调用被拒绝：{reason}")
+        except Exception:  # noqa: BLE001
+            log.warning("回写拒绝消息失败 conv=%s", conv_id)
+        return {"ok": False, "reason": reason}
+
+    # 在线查询
+    online_info: dict = {}
+    try:
+        online_info = await im.internal_agent_online(target_addr)
+    except Exception:  # noqa: BLE001
+        log.warning("agent 在线查询失败 addr=%s", target_addr)
+    online = bool(online_info.get("online"))
+
+    if not online:
+        try:
+            await im.internal_agent_reply(
+                conv_id, "", "offline",
+                f"成员智能体当前不在线，无法执行（目标：{target_addr}）",
+            )
+        except Exception:  # noqa: BLE001
+            log.warning("回写离线消息失败 conv=%s", conv_id)
+        return {"ok": False, "reason": "offline"}
+
+    # 组装 invoke payload（复用 inter-agent 协议）
+    payload = {
+        "v": 1,
+        "caller": {
+            "user_id": sender_id,
+            "agent_id": metadata.get("caller_agent_id", ""),
+            "name": sender_name or sender_id,
+        },
+        "session_id": metadata.get("session_id", "") or "",
+        "prompt": prompt,
+        "context": {
+            "root_session_id": metadata.get("root_session_id", ""),
+            "company_id": metadata.get("company_id", ""),
+        },
+        "mode": metadata.get("mode", "auto"),
+    }
+
+    try:
+        result = await im.internal_agent_invoke(target_addr, payload, conv_id)
+    except Exception as exc:  # noqa: BLE001
+        log.warning("agent invoke 派发失败 addr=%s: %s", target_addr, exc)
+        try:
+            await im.internal_agent_reply(conv_id, "", "error", f"任务派发失败：{exc}")
+        except Exception:  # noqa: BLE001
+            pass
+        return {"ok": False, "reason": str(exc)}
+
+    return {"ok": True, "task_id": result.get("task_id", "")}

app/api/routers/agent_gate.py

@@ -0,0 +1,106 @@
+# -*- coding: utf-8 -*-
+"""智能体互通 API：内部事件接收 + 临时授权管理。
+
+- ``POST /internal/im/agent-events``：im-service 在 agent_task 消息落库后回调，
+  由编排层完成权限校验 → 在线查询 → MQTT 派发（携带 X-IM-Token 内部令牌）。
+- ``POST/GET/DELETE /agent-grants``：L3 成员开放调用（临时授权）管理（用户 API）。
+"""
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, Depends, Header, HTTPException
+
+from ..dependencies import get_db, get_current_user
+from ... import config
+from ...agent_gate import handle_agent_task_event
+from ...im import client as im_client
+from ...infrastructure.repositories import Database
+
+log = logging.getLogger("agent_gate.api")
+
+router = APIRouter(tags=["agent-gate"])
+internal_router = APIRouter(prefix="/internal/im", tags=["agent-gate-internal"])
+
+
+# ── 内部事件接收（im-service → server-core） ─────────────────────
+
+def _require_im_internal(x_im_token: str = Header(default="", alias="X-IM-Token")) -> None:
+    """仅允许携带 IM_INTERNAL_TOKEN 的调用方（im-service）。"""
+    if not config.IM_INTERNAL_TOKEN or x_im_token != config.IM_INTERNAL_TOKEN:
+        raise HTTPException(status_code=403, detail="内部接口访问被拒绝")
+
+
+@internal_router.post("/agent-events", dependencies=[Depends(_require_im_internal)])
+async def agent_events(payload: dict, db: Database = Depends(get_db)):
+    """接收 im-service 的智能体任务事件（agent_task 消息落库后回调）。
+
+    body: {event, conv_id, sender_id, sender_name, content, metadata}
+    """
+    event = payload.get("event", "")
+    if event != "agent_task":
+        return {"ok": False, "reason": f"unsupported event: {event}"}
+    conv_id = str(payload.get("conv_id", ""))
+    if not conv_id:
+        return {"ok": False, "reason": "missing conv_id"}
+    result = await handle_agent_task_event(
+        db,
+        im_client,
+        conv_id=conv_id,
+        sender_id=str(payload.get("sender_id", "")),
+        sender_name=str(payload.get("sender_name", "")),
+        content=str(payload.get("content", "")),
+        metadata=payload.get("metadata") or {},
+    )
+    return result
+
+
+# ── 用户 API：L3 临时授权管理 ────────────────────────────────────
+
+@router.post("/agent-grants", summary="创建智能体临时授权")
+async def create_grant(payload: dict, user: dict = Depends(get_current_user),
+                       db: Database = Depends(get_db)):
+    """成员开放调用：授权另一个成员调用自己（的某个）智能体。
+
+    body: {grantee_user_id, target_agent_addr, scope?, max_calls?, expires_at?}
+    """
+    grantee = str(payload.get("grantee_user_id", "")).strip()
+    target_addr = str(payload.get("target_agent_addr", "")).strip()
+    if not grantee or not target_addr:
+        raise HTTPException(status_code=422, detail="grantee_user_id 与 target_agent_addr 必填")
+
+    from ...agent_gate import parse_agent_addr
+    parsed = parse_agent_addr(target_addr)
+    if parsed is None or parsed[0] != user["id"]:
+        raise HTTPException(status_code=403, detail="只能授权自己名下的智能体")
+    if grantee == user["id"]:
+        raise HTTPException(status_code=422, detail="不能授权给自己")
+
+    grant = await db.agent_access_grants.create(
+        grantor_user_id=user["id"],
+        target_agent_addr=target_addr,
+        grantee_user_id=grantee,
+        scope=str(payload.get("scope", "")),
+        max_calls=int(payload.get("max_calls", 0) or 0),
+        expires_at=str(payload.get("expires_at", "")),
+    )
+    return grant
+
+
+@router.get("/agent-grants", summary="智能体临时授权列表")
+async def list_grants(direction: str = "grantor",
+                      user: dict = Depends(get_current_user),
+                      db: Database = Depends(get_db)):
+    """direction=grantor（我授权的）| grantee（我被授权的）。"""
+    if direction == "grantee":
+        return await db.agent_access_grants.list_by_grantee(user["id"])
+    return await db.agent_access_grants.list_by_grantor(user["id"])
+
+
+@router.delete("/agent-grants/{grant_id}", summary="撤销智能体临时授权")
+async def revoke_grant(grant_id: str, user: dict = Depends(get_current_user),
+                       db: Database = Depends(get_db)):
+    ok = await db.agent_access_grants.revoke(grant_id, user["id"])
+    if not ok:
+        raise HTTPException(status_code=404, detail="授权不存在或无权撤销")
+    return {"ok": True}

app/im/client.py

@@ -130,6 +130,45 @@ async def internal_mqtt_credentials(user_id: str) -> dict:
     )
 
 
+async def internal_agent_invoke(addr: str, payload: dict, conv_id: str = "") -> dict:
+    """派发智能体调用任务：im-service 发布 MQTT invoke 并登记 task。
+
+    - addr：目标 agent 地址 ``{user_id}.{agent_id}``
+    - payload：invoke 协议（caller/session_id/prompt/mode/...）
+    - conv_id：回写目标会话（agent_reply 落库到该会话）
+    """
+    return await _request(
+        method="POST", path="/internal/agent/invoke",
+        headers=_internal_headers(),
+        json_body={"addr": str(addr), "payload": payload, "conv_id": str(conv_id or "")},
+    )
+
+
+async def internal_agent_online(addr: str) -> dict:
+    """查询目标 agent 是否在线（EMQX presence / 客户端列表）。"""
+    return await _request(
+        method="GET", path="/internal/agent/online",
+        headers=_internal_headers(),
+        params={"addr": str(addr)},
+    )
+
+
+async def internal_agent_reply(conv_id: str, task_id: str, status: str,
+                               content: str, sender: dict | None = None) -> dict:
+    """向会话写入 agent_reply 消息（离线/失败回执或编排层主动回复）。"""
+    return await _request(
+        method="POST", path="/internal/agent/reply",
+        headers=_internal_headers(),
+        json_body={
+            "conv_id": str(conv_id),
+            "task_id": str(task_id or ""),
+            "status": str(status),
+            "content": str(content),
+            "sender": sender or {},
+        },
+    )
+
+
 async def health() -> dict:
     try:
         return await _request(method="GET", path="/health")

app/infrastructure/models.py

@@ -13,6 +13,7 @@ from sqlalchemy import (
     Boolean,
     Float,
     ForeignKey,
+    Index,
     Integer,
     PrimaryKeyConstraint,
     String,
@@ -208,6 +209,36 @@ class Agent(Base):
     model_name: Mapped[str] = mapped_column(String, default="")
     deletable: Mapped[bool] = mapped_column(Boolean, default=True)
     use_fixed_soul: Mapped[bool] = mapped_column(Boolean, default=False)
+    external_callable: Mapped[bool] = mapped_column(Boolean, default=True)   # 是否允许被企业内其他账号/智能体调用
+    call_mode: Mapped[str] = mapped_column(String, default="auto")           # auto=自动执行 / confirm=需归属用户确认
+    created_at: Mapped[str] = mapped_column(String, default="")
+    updated_at: Mapped[str] = mapped_column(String, default="")
+
+
+class AgentAccessGrant(Base):
+    """智能体临时开放调用授权（成员 A 授权成员 B 调用 A 的智能体）。
+
+    三层权限模型（集中校验于 server-core）：
+      L1 企业管理员 → 本企业任意成员 agent（无需授权记录）
+      L2 同一企业内 agent 互调（企业=信任域）
+      L3 本表：跨成员显式授权（限时/限次/限 scope）
+    """
+
+    __tablename__ = "agent_access_grants"
+    __table_args__ = (
+        Index("idx_agent_grant_grantee", "grantee_user_id", "status"),
+        Index("idx_agent_grant_target", "target_agent_addr", "status"),
+    )
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    grantor_user_id: Mapped[str] = mapped_column(String, index=True)        # 授权人（agent 归属用户）
+    target_agent_addr: Mapped[str] = mapped_column(String, index=True)      # 目标 agent 地址 {user_id}.{agent_id}
+    grantee_user_id: Mapped[str] = mapped_column(String, index=True)        # 被授权人（可调用方）
+    scope: Mapped[str] = mapped_column(Text, default="")                    # prompt 前缀白名单（每行一条前缀），空=全部
+    max_calls: Mapped[int] = mapped_column(Integer, default=0)              # 0=不限次数
+    used_calls: Mapped[int] = mapped_column(Integer, default=0)
+    expires_at: Mapped[str] = mapped_column(String, default="")             # ISO 时间（UTC）；空=永久
+    status: Mapped[str] = mapped_column(String, default="active")           # active|revoked
     created_at: Mapped[str] = mapped_column(String, default="")
     updated_at: Mapped[str] = mapped_column(String, default="")
 

app/infrastructure/repositories.py

@@ -1106,6 +1106,8 @@ class AgentRepository:
             "model_name": a.model_name,
             "deletable": a.deletable,
             "use_fixed_soul": a.use_fixed_soul,
+            "external_callable": bool(a.external_callable),
+            "call_mode": a.call_mode,
             "created_at": a.created_at,
             "updated_at": a.updated_at,
         }
@@ -1140,6 +1142,8 @@ class AgentRepository:
         model_name: str = "",
         agent_id: str | None = None,
         port: str = "opc",
+        external_callable: bool = True,
+        call_mode: str = "auto",
     ) -> dict:
         now = utcnow_iso()
         a = Agent(
@@ -1152,6 +1156,8 @@ class AgentRepository:
             model_name=model_name,
             deletable=True,
             use_fixed_soul=False,
+            external_callable=external_callable,
+            call_mode=call_mode or "auto",
             created_at=now,
             updated_at=now,
         )
@@ -1169,7 +1175,8 @@ class AgentRepository:
         allowed = {
             k: fields[k]
             for k in fields
-            if k in ("name", "description", "language", "model_name", "use_fixed_soul")
+            if k in ("name", "description", "language", "model_name", "use_fixed_soul",
+                     "external_callable", "call_mode")
         }
         if not allowed:
             return None
@@ -1191,6 +1198,105 @@ class AgentRepository:
         return True
 
 
+# ---------------------------------------------------------------------------
+# 智能体临时开放调用授权（L3：成员间显式授权）
+# ---------------------------------------------------------------------------
+class AgentAccessGrantRepository:
+    """agent_access_grants：成员 A 临时授权成员 B 调用 A 的智能体。
+
+    调用方身份：grantee_user_id（可调用方用户）。
+    校验逻辑（server-core 编排层）：L1 管理员无条件 → L2 同企业互调 → L3 本表。
+    """
+
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    @staticmethod
+    def _to_dict(g: AgentAccessGrant) -> dict:
+        return {
+            "id": g.id,
+            "grantor_user_id": g.grantor_user_id,
+            "target_agent_addr": g.target_agent_addr,
+            "grantee_user_id": g.grantee_user_id,
+            "scope": g.scope,
+            "max_calls": g.max_calls,
+            "used_calls": g.used_calls,
+            "expires_at": g.expires_at,
+            "status": g.status,
+            "created_at": g.created_at,
+            "updated_at": g.updated_at,
+        }
+
+    async def create(
+        self,
+        grantor_user_id: str,
+        target_agent_addr: str,
+        grantee_user_id: str,
+        *,
+        scope: str = "",
+        max_calls: int = 0,
+        expires_at: str = "",
+    ) -> dict:
+        now = utcnow_iso()
+        g = AgentAccessGrant(
+            id=new_id("grant"),
+            grantor_user_id=grantor_user_id,
+            target_agent_addr=target_agent_addr,
+            grantee_user_id=grantee_user_id,
+            scope=scope or "",
+            max_calls=max(0, int(max_calls or 0)),
+            used_calls=0,
+            expires_at=expires_at or "",
+            status="active",
+            created_at=now,
+            updated_at=now,
+        )
+        self.session.add(g)
+        await self.session.commit()
+        return self._to_dict(g)
+
+    async def list_by_grantor(self, grantor_user_id: str, status: str = "active") -> list[dict]:
+        stmt = select(AgentAccessGrant).where(
+            AgentAccessGrant.grantor_user_id == grantor_user_id,
+        ).order_by(AgentAccessGrant.created_at.desc())
+        rows = await self.session.scalars(stmt)
+        return [self._to_dict(g) for g in rows if g.status == status or status == "all"]
+
+    async def list_by_grantee(self, grantee_user_id: str, status: str = "active") -> list[dict]:
+        stmt = select(AgentAccessGrant).where(
+            AgentAccessGrant.grantee_user_id == grantee_user_id,
+        ).order_by(AgentAccessGrant.created_at.desc())
+        rows = await self.session.scalars(stmt)
+        return [self._to_dict(g) for g in rows if g.status == status or status == "all"]
+
+    async def get_active(
+        self, grantor_user_id: str, target_agent_addr: str, grantee_user_id: str,
+    ) -> AgentAccessGrant | None:
+        """取一条有效授权（用于调用校验）。"""
+        stmt = select(AgentAccessGrant).where(
+            AgentAccessGrant.grantor_user_id == grantor_user_id,
+            AgentAccessGrant.target_agent_addr == target_agent_addr,
+            AgentAccessGrant.grantee_user_id == grantee_user_id,
+            AgentAccessGrant.status == "active",
+        )
+        return await self.session.scalar(stmt)
+
+    async def revoke(self, grant_id: str, grantor_user_id: str) -> bool:
+        g = await self.session.get(AgentAccessGrant, grant_id)
+        if g is None or g.grantor_user_id != grantor_user_id:
+            return False
+        g.status = "revoked"
+        g.updated_at = utcnow_iso()
+        await self.session.commit()
+        return True
+
+    async def consume_call(self, grant: AgentAccessGrant) -> None:
+        """调用成功后计数（限次授权）。"""
+        grant.used_calls = (grant.used_calls or 0) + 1
+        grant.updated_at = utcnow_iso()
+        await self.session.commit()
+
+
 # ---------------------------------------------------------------------------
 # 官方预置智能体配置（管理后台可配置，替代硬编码 AGENT_SEED）
 # ---------------------------------------------------------------------------
@@ -3282,6 +3388,7 @@ class Database:
         self.tokens = TokenRepository(self.session)
         self.agents = AgentRepository(self.session)
         self.official_agents = OfficialAgentRepository(self.session)
+        self.agent_access_grants = AgentAccessGrantRepository(self.session)
         self.roles = RoleRepository(self.session)
         self.orgs = OrgRepository(self.session)
         self.regions = RegionRepository(self.session)

app/main.py

@@ -43,6 +43,7 @@ from app.market import routers as market_router
 from app.incubator import routers as incubator_router
 from app.api.routers import invite as invite_router
 from app.api.routers import rbac_compute_pricing as compute_pricing_router
+from app.api.routers import agent_gate as agent_gate_router
 from app.im import router as im_router
 
 APP_NAME = "云南省超级个体服务平台"
@@ -132,6 +133,8 @@ app.include_router(market_router.router)
 app.include_router(market_router.admin_router)
 app.include_router(incubator_router.router)
 app.include_router(invite_router.router)
+app.include_router(agent_gate_router.router)
+app.include_router(agent_gate_router.internal_router)
 app.include_router(im_router.router)
 
 

scripts/migrate_agent_gate.sql

@@ -0,0 +1,63 @@
+-- ============================================================
+-- 智能体互通（agent_gate）数据库迁移
+-- 适用：server-core 生产库（本机与生产共用同一 MySQL）
+-- 执行方式（本机）：
+--   mysql -h47.108.226.213 -P8091 -uopc -pjjjsgysyujkwjwgb opc < migrate_agent_gate.sql
+-- 幂等：全部语句可重复执行（先检查列/表是否已存在）
+-- ============================================================
+
+-- 1) agents 表新增：外部调用开关 + 调用确认模式
+SET @col_exists := (
+  SELECT COUNT(*) FROM information_schema.COLUMNS
+  WHERE TABLE_SCHEMA = 'opc' AND TABLE_NAME = 'agents' AND COLUMN_NAME = 'external_callable'
+);
+SET @sql := IF(@col_exists = 0,
+  'ALTER TABLE agents ADD COLUMN external_callable TINYINT(1) NOT NULL DEFAULT 1 COMMENT ''是否允许被企业内其他账号/智能体调用''',
+  'SELECT 1');
+PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
+
+SET @col_exists := (
+  SELECT COUNT(*) FROM information_schema.COLUMNS
+  WHERE TABLE_SCHEMA = 'opc' AND TABLE_NAME = 'agents' AND COLUMN_NAME = 'call_mode'
+);
+SET @sql := IF(@col_exists = 0,
+  'ALTER TABLE agents ADD COLUMN call_mode VARCHAR(16) NOT NULL DEFAULT ''auto'' COMMENT ''auto=自动执行 / confirm=需归属用户确认''',
+  'SELECT 1');
+PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
+
+-- 2) 智能体临时授权表（L3 成员开放调用）
+CREATE TABLE IF NOT EXISTS agent_access_grants (
+  id                  VARCHAR(64)  NOT NULL COMMENT 'grant_xxx',
+  grantor_user_id     VARCHAR(36)  NOT NULL COMMENT '授权人（agent 归属用户）',
+  target_agent_addr   VARCHAR(128) NOT NULL COMMENT '目标 agent 地址 {user_id}.{agent_id}',
+  grantee_user_id     VARCHAR(36)  NOT NULL COMMENT '被授权人（可调用方）',
+  scope               TEXT         NOT NULL COMMENT 'prompt 前缀白名单（每行一条），空=全部',
+  max_calls           INT          NOT NULL DEFAULT 0 COMMENT '0=不限次数',
+  used_calls          INT          NOT NULL DEFAULT 0,
+  expires_at          VARCHAR(32)  NOT NULL DEFAULT '' COMMENT 'ISO 时间(UTC)；空=永久',
+  status              VARCHAR(16)  NOT NULL DEFAULT 'active' COMMENT 'active|revoked',
+  created_at          VARCHAR(32)  NOT NULL DEFAULT '',
+  updated_at          VARCHAR(32)  NOT NULL DEFAULT '',
+  PRIMARY KEY (id),
+  KEY idx_agent_grant_grantee (grantee_user_id, status),
+  KEY idx_agent_grant_target (target_agent_addr, status),
+  KEY idx_agent_grant_grantor (grantor_user_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='智能体临时开放调用授权（L3）';
+
+-- 3) im-service 智能体任务登记表（im_agent_tasks，幂等兜底；im-service 启动时也会 create_all）
+CREATE TABLE IF NOT EXISTS im_agent_tasks (
+  task_id           VARCHAR(48)  NOT NULL COMMENT 'agt_xxx',
+  addr              VARCHAR(128) NOT NULL DEFAULT '' COMMENT '{user_id}.{agent_id}',
+  conv_id           VARCHAR(96)  NOT NULL DEFAULT '' COMMENT 'agent_reply 回写会话',
+  caller_user_id    VARCHAR(36)  NOT NULL DEFAULT '',
+  caller_name       VARCHAR(64)  NOT NULL DEFAULT '',
+  prompt            TEXT,
+  mode              VARCHAR(16)  NOT NULL DEFAULT 'auto' COMMENT 'auto|confirm',
+  status            VARCHAR(16)  NOT NULL DEFAULT 'pending' COMMENT 'pending|running|ok|error|offline|cancelled',
+  reply_content     TEXT,
+  created_at        VARCHAR(32)  NOT NULL DEFAULT '',
+  updated_at        VARCHAR(32)  NOT NULL DEFAULT '',
+  PRIMARY KEY (task_id),
+  KEY idx_im_agent_task_conv (conv_id),
+  KEY idx_im_agent_task_addr (addr, status)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='智能体互通任务登记（im-service）';