Align docs and runtime with expert-only routing.

Drop dead Manager helpers from the gateway, fix async wire_policy, and remove stale comprehensive UI copy from RUNBOOK/README/presentation.

README.md

@@ -111,7 +111,7 @@ powershell -ExecutionPolicy Bypass -File .\runtime\operations\scripts\weixin_sta
 Notes:
 - `weixin_install.ps1` does **not** require global `openclaw` CLI installation; runtime deps are installed locally in sidecar workspace.
 - `.\scripts\start_all.ps1 -Background` now skips missing Weixin/WhatsApp sidecars gracefully (warn + continue), so Admin/Chat can still boot on fresh installs.
-- Admin supports channel dispatch controls for Weixin/WhatsApp (bind specialist / comprehensive), with default `generalist`.
+- Admin supports channel dispatch controls for Weixin/WhatsApp (bind default specialist in expert mode), with default `generalist`.
 
 Full runbook (recommended): see `docs/RUNBOOK.md` → “开源快速安装（从零到跑起来）”.
 WhatsApp install/start guide: see `docs/RUNBOOK.md` → “4.2 WhatsApp（实验接入）”.

docs/RUNBOOK.md

@@ -243,16 +243,15 @@ Linux/macOS:
 Admin 可视化调度（新增）：
 
 - `Stack` 页面提供 `Weixin dispatch` 控制卡。
-- 可直接选择专家并执行：
-  - `绑定专家`（写入 `expert + specialist`）
-  - `综合`（写入 `comprehensive + specialist`）
+- 可直接选择专家并执行 `绑定专家`（写入 `expert + specialist`）。
 - 通道默认值：`expert + generalist`。
+- 产品面已统一为**专家模式**；不再提供「综合 / Manager」分派。
 
 账号级调度（新增）：
 
 - 在 `用户/渠道绑定` 页面选择 `channel=weixin` 后，可按账号配置：
   - 专家（默认 `generalist`）
-  - 模式：`绑定专家` / `综合`
+  - 绑定专家（`expert + specialist`）
 - 生效优先级：**账号级配置 > 通道全局配置 > 默认值**。
 
 注意：当模型侧返回“OpenAI key 缺失”兜底文本时，微信通道会静默抑制该类回复（不向微信用户下发错误文案）。
@@ -310,9 +309,9 @@ powershell -ExecutionPolicy Bypass -File .\runtime\operations\scripts\whatsapp_s
 
 Admin 可视化调度（新增）：
 
-- `Stack` 页面提供 `WhatsApp dispatch` 控制卡（`绑定专家` / `综合`）。
-- `用户/渠道绑定` 页面选择 `channel=whatsapp` 后可按账号单独配置专家/模式。
-- 默认值同微信：`expert + generalist`。
+- `Stack` 页面提供 `WhatsApp dispatch` 控制卡（`绑定专家`）。
+- `用户/渠道绑定` 页面选择 `channel=whatsapp` 后可按账号单独配置默认专家。
+- 默认值同微信：`expert + generalist`（专家模式 only）。
 
 ---
 
@@ -403,7 +402,7 @@ Admin Chat 依赖表 **`ui_session_owner`**（`session_id` → `tenant_id` + `us
 
 **工作区 ``extra_roots``（与编排临时会话）**
 
-管理台为用户配置的 ``user_workspace_path_allowlist.extra_roots`` 通过 ``ui_session_owner`` 解析到租户+用户。全能者编排里专家步往往在**无 owner 的临时** ``chat_session`` 上落库中间消息：内置路径类工具会携带 **用户 UI 会话 id 作为 fallback**，仍按该用户策略合并 ``extra_roots``；MCP filesystem 启动参数本就按用户聊天 ``session_id``（policy）合并，二者现已对齐。
+管理台为用户配置的 ``user_workspace_path_allowlist.extra_roots`` 通过 ``ui_session_owner`` 解析到租户+用户。专家轮次若在**无 owner 的临时** ``chat_session`` 上落库中间消息：内置路径类工具会携带 **用户 UI 会话 id 作为 fallback**，仍按该用户策略合并 ``extra_roots``；MCP filesystem 启动参数本就按用户聊天 ``session_id``（policy）合并，二者现已对齐。
 
 ### 6.5 主库路径与「删掉的会话又回来了 / 新建用户不见了」
 

docs/presentations/oclaw-netx-ops-expert.md

@@ -64,12 +64,11 @@ flowchart LR
 | **ops** | 网络运维专家（本文重点） |
 | generalist | 通用任务 |
 | memory | 知识/记忆 |
-| image / video / stock | 专项能力 |
 
 ### 2.3 交互模式
 
-- **expert**：用户直连某一专家（如 ops）
-- **comprehensive**：经理 Agent 编排，按任务委派给 ops 等专家
+- 产品面已统一为 **expert**：用户直连某一专家（如 ops）
+- 历史「comprehensive / Manager 编排」已移除
 
 ### 2.4 分层架构
 
@@ -87,8 +86,6 @@ skills/         可安装技能包（SKILL.md）
 flowchart TD
   A[入站消息] --> B{渠道路由}
   B -->|specialist=ops| C[NetworkOpsAgent]
-  B -->|comprehensive| D[Manager Agent]
-  D -->|委派网络任务| C
   C --> E[加载 ROLE_SYSTEM + Skills]
   E --> F[工具集: network_ops + memory + MCP]
   F --> G{Agent 循环}
@@ -448,8 +445,7 @@ powershell -ExecutionPolicy Bypass -File .\runtime\operations\scripts\whatsapp_s
 
 **交互模式：**
 
-- **绑定专家（expert）**：所有消息直连 ops 运维专家（推荐运维群场景）
-- **综合（comprehensive）**：经理 Agent 编排，网络任务委派给 ops
+- **绑定专家（expert）**：所有消息直连 ops 运维专家（推荐运维群场景；当前唯一模式）
 
 **优先级**：账号级配置 > 通道全局配置 > 默认（`expert + generalist`）
 

runtime/gateway.py

@@ -18,7 +18,6 @@ from runtime.hooks_runtime import (
 )
 from runtime.relay_pointer import summarize_relay_ttl
 from runtime.skills import build_skill_manifest
-from runtime.prompt_prebuild import get_manager_prompt_prebuild
 from runtime.types import (
     OclawSessionContext,
     StandardMessage,
@@ -120,71 +119,6 @@ class OclawGateway:
         except Exception:
             pass
 
-    @staticmethod
-    def _looks_like_manager_instruction(reply: str, instruction: str) -> bool:
-        r = str(reply or "").strip()
-        ins = str(instruction or "").strip()
-        if not r or not ins:
-            return False
-        if r == ins:
-            return True
-        if len(ins) >= 16 and ins in r:
-            return True
-        # Heuristic: prefix overlap is usually enough for instruction leakage.
-        a = r[:120]
-        b = ins[:120]
-        common = 0
-        for x, y in zip(a, b):
-            if x != y:
-                break
-            common += 1
-        return common >= 24
-
-    @staticmethod
-    def _parse_json_object(text: str) -> dict[str, Any] | None:
-        t = str(text or "").strip()
-        start = t.find("{")
-        if start < 0:
-            return None
-        executed_turn_uuid = ""
-        try:
-            obj, _end = json.JSONDecoder().raw_decode(t[start:])
-        except Exception:
-            return None
-        return obj if isinstance(obj, dict) else None
-
-    @staticmethod
-    def _sanitize_dynamic_system_prompt(raw: Any) -> str:
-        s = str(raw or "").strip()
-        if not s:
-            return ""
-        s = s[:3000]
-        banned = ("<tool_call>", "</tool_call>", "assistant_response:", "function_call:")
-        low = s.lower()
-        if any(b in low for b in banned):
-            return ""
-        return s
-
-    @staticmethod
-    def _parse_dynamic_agent(raw: Any) -> dict[str, Any] | None:
-        if not isinstance(raw, dict):
-            return None
-        name = str(raw.get("name") or "").strip()
-        system_prompt = OclawGateway._sanitize_dynamic_system_prompt(raw.get("system_prompt"))
-        reason = str(raw.get("reason") or "").strip()
-        tp = raw.get("tool_policy")
-        tool_policy = tp if isinstance(tp, dict) else {}
-        allow_tags = [str(x).strip() for x in (tool_policy.get("allow_tags") or []) if str(x).strip()]
-        allow_tools = [str(x).strip() for x in (tool_policy.get("allow_tools") or []) if str(x).strip()]
-        if not system_prompt:
-            return None
-        return {
-            "name": name or "dynamic_ephemeral",
-            "system_prompt": system_prompt,
-            "tool_policy": {"allow_tags": allow_tags, "allow_tools": allow_tools},
-            "reason": reason or "dynamic_agent_selected",
-        }
-
     def _maybe_generate_title_on_third_round(self, *, msg: StandardMessage, model: Any | None) -> None:
         """Generate title once on round-3: one plain model.chat (system+user, no tools)."""
         if model is None or not callable(getattr(model, "chat", None)):
@@ -299,123 +233,6 @@ class OclawGateway:
         except Exception:
             pass
 
-    def _manager_select_specialist(
-        self,
-        *,
-        msg: StandardMessage,
-        lang: str,
-        executor: Any,
-        memory_enabled: bool,
-    ) -> tuple[str, str, dict[str, Any] | None, str]:
-        model = getattr(executor, "model", None)
-        if model is None or not callable(getattr(model, "chat", None)):
-            return ("generalist", "manager_model_missing", None, "")
-        try:
-            registry = getattr(executor, "tools", None)
-            base_url = str(getattr(model, "base_url", "") or "")
-            if registry is None:
-                return ("generalist", "manager_tools_missing", None, "")
-            pack = get_manager_prompt_prebuild(
-                store=self.store,
-                registry=registry,
-                base_url=base_url,
-                memory_enabled=memory_enabled,
-            )
-            manager_context = str(pack.get("manager_context") or "")
-            allowed_fixed = [str(x).strip().lower() for x in (pack.get("allowed_fixed") or []) if str(x).strip()]
-            allowed_fixed_quoted = str(pack.get("allowed_fixed_quoted") or "")
-            messages = [
-                {
-                    "role": "system",
-                    "content": (
-                        f"{manager_context}\n\n"
-                        "Return exactly one compact JSON object with route.specialist, route.reason, and "
-                        "dispatch.instruction_text. "
-                        f"Allowed fixed specialists: {allowed_fixed_quoted}. "
-                        "If route.specialist is NOT a fixed specialist, you MUST include dynamic_agent with "
-                        "name/system_prompt/tool_policy(allow_tags/allow_tools)/reason."
-                    ),
-                },
-                {
-                    "role": "user",
-                    "content": (
-                        f"User request:\n{str(msg.text or '').strip()}\n\n"
-                        "Return JSON only."
-                    ),
-                },
-            ]
-            ensure_no_tool_or_embedded_image_payload(messages=messages, path="gateway.manager_select")
-            resp = model.chat(messages, [], on_token=None)
-            obj = self._parse_json_object(str(getattr(resp, "content", "") or ""))
-            if not isinstance(obj, dict):
-                return ("generalist", "manager_json_missing", None, "")
-            route = obj.get("route") if isinstance(obj, dict) else None
-            if not isinstance(route, dict):
-                return ("generalist", "manager_route_missing", None, "")
-            route_kind = str(route.get("kind") or "").strip().lower()
-            raw_specialist = str(route.get("specialist") or "").strip().lower()
-            fixed_set = set([str(x).strip().lower() for x in allowed_fixed if str(x).strip()])
-            fixed = raw_specialist in fixed_set
-            specialist = normalize_requested_specialist(raw_specialist) if fixed else raw_specialist
-            reason = str(route.get("reason") or "").strip() or "manager_selected"
-            dispatch = obj.get("dispatch") if isinstance(obj, dict) else None
-            instruction_text = ""
-            if isinstance(dispatch, dict):
-                instruction_text = str(dispatch.get("instruction_text") or "").strip()
-            if not instruction_text:
-                return ("generalist", "manager_instruction_missing", None, "")
-            if route_kind and route_kind != "specialist":
-                return ("generalist", "manager_route_kind_invalid", None, instruction_text)
-            dynamic_agent = self._parse_dynamic_agent(obj.get("dynamic_agent") if isinstance(obj, dict) else None)
-            if specialist == "memory" and not memory_enabled:
-                return ("generalist", "memory_disabled_fallback", None, instruction_text)
-            if not fixed and dynamic_agent is None:
-                return ("generalist", "dynamic_agent_invalid_fallback", None, instruction_text)
-            return (specialist, reason, dynamic_agent, instruction_text)
-        except Exception:
-            return ("generalist", "manager_select_failed", None, "")
-
-    def _manager_finalize_output(
-        self,
-        *,
-        msg: StandardMessage,
-        lang: str,
-        executor: Any,
-        specialist: str,
-        specialist_reply: str,
-        memory_enabled: bool,
-        on_token: Optional[Callable[[str], None]] = None,
-    ) -> str:
-        model = getattr(executor, "model", None)
-        if model is None or not callable(getattr(model, "chat", None)):
-            return str(specialist_reply or "")
-        try:
-            registry = getattr(executor, "tools", None)
-            base_url = str(getattr(model, "base_url", "") or "")
-            if registry is None:
-                return str(specialist_reply or "")
-            pack = get_manager_prompt_prebuild(
-                store=self.store,
-                registry=registry,
-                base_url=base_url,
-                memory_enabled=memory_enabled,
-            )
-            manager_context = str(pack.get("manager_context") or "")
-            user_text = (
-                "请基于以下信息输出最终答复。\n\n"
-                f"原始用户问题:\n{str(msg.text or '').strip()}\n\n"
-                f"已调用专家: {str(specialist or '').strip()}\n\n"
-                f"专家结果:\n{str(specialist_reply or '').strip()}\n\n"
-                "要求：保持简洁、准确，不要暴露内部流程。"
-            )
-            messages = [{"role": "system", "content": manager_context}, {"role": "user", "content": user_text}]
-            ensure_no_tool_or_embedded_image_payload(messages=messages, path="gateway.manager_finalize")
-            resp = model.chat(messages, [], on_token=on_token)
-            final_text = str(getattr(resp, "content", "") or "").strip()
-            return final_text or str(specialist_reply or "")
-        except Exception:
-            return str(specialist_reply or "")
-
     def _memory_enabled(self) -> bool:
         raw = str(self.store.get_setting(_SPECIALIST_FLAGS_SETTING_KEY) or "").strip()
         if not raw:

runtime/worker.py

@@ -254,10 +254,8 @@ def _worker_loop(*, store: Any, worker_id: str, poll_interval_s: float) -> None:
                     or ""
                 )
             )
-            skill_binding_role = str(manager_specialist or "generalist")
-            wire_policy_role = (
-                "manager" if interaction_mode == "comprehensive" else str(requested_specialist or skill_binding_role)
-            )
+            skill_binding_role = str(manager_specialist or requested_specialist or "generalist")
+            wire_policy_role = str(requested_specialist or skill_binding_role)
             memory_ctx = build_memory_context(
                 store=store,
                 session_id=session_id,