4344 lines
168 KiB
Python
4344 lines
168 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Project-local integration layer for xai-org/grok-build.
|
||
|
||
The upstream Grok Build binary remains the execution engine. This module owns
|
||
the project-scoped runtime state, installs the official Windows release, maps
|
||
the backend-managed model into Grok's TOML format without persisting its API
|
||
key in TOML, and exposes interactive, headless, and ACP launch modes.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import copy
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import platform
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import tomllib
|
||
import urllib.error
|
||
import urllib.request
|
||
import uuid
|
||
from datetime import date, datetime, time as datetime_time
|
||
from dataclasses import asdict, dataclass, replace
|
||
from pathlib import Path
|
||
from typing import Callable, Iterable, Mapping, Sequence
|
||
from urllib.parse import parse_qsl, urlsplit, urlunsplit
|
||
|
||
|
||
PROJECT_DIR = Path(__file__).resolve().parent
|
||
DEFAULT_AI_SETTINGS = PROJECT_DIR / "ai_settings.local.json"
|
||
DEFAULT_INTEGRATION_SETTINGS = PROJECT_DIR / "grok_build_settings.json"
|
||
OFFICIAL_BASE_URL = "https://x.ai/cli"
|
||
OFFICIAL_FALLBACK_URL = "https://storage.googleapis.com/grok-build-public-artifacts/cli"
|
||
MODEL_PROFILE = "wecom-backend"
|
||
MODEL_API_KEY_ENV = "WECOM_GROK_API_KEY"
|
||
MANAGED_MODELS_KEYS = frozenset(
|
||
{
|
||
"default",
|
||
"allowed_models",
|
||
"web_search",
|
||
"session_summary",
|
||
"image_description",
|
||
"prompt_suggestion",
|
||
}
|
||
)
|
||
MANAGED_UI_KEYS = frozenset({"prompt_suggestions", "fork_secondary_model"})
|
||
PINNED_SUBAGENT_NAMES = frozenset(
|
||
{"general-purpose", "explore", "plan"}
|
||
)
|
||
CUSTOM_MODEL_ENVIRONMENT = {
|
||
"GROK_DEFAULT_MODEL": MODEL_PROFILE,
|
||
"GROK_WEB_SEARCH_MODEL": MODEL_PROFILE,
|
||
"GROK_SESSION_SUMMARY_MODEL": MODEL_PROFILE,
|
||
"GROK_IMAGE_DESCRIPTION_MODEL": MODEL_PROFILE,
|
||
"GROK_PROMPT_SUGGESTIONS_MODEL": MODEL_PROFILE,
|
||
"GROK_SUGGESTIONS_AI_MODEL": MODEL_PROFILE,
|
||
"GROK_GOAL_USE_CURRENT_MODEL_ONLY": "1",
|
||
"GROK_PROMPT_SUGGESTIONS": "0",
|
||
"GROK_SUGGESTIONS": "0",
|
||
"GROK_SUGGESTIONS_AI": "0",
|
||
"GROK_AGENT": "grok-build",
|
||
"GROK_MEMORY": "0",
|
||
"GROK_IMAGE_GEN": "0",
|
||
"GROK_IMAGE_EDIT": "0",
|
||
"GROK_VIDEO_GEN": "0",
|
||
}
|
||
XAI_CREDENTIAL_ENV_VARS = frozenset(
|
||
{
|
||
"XAI_API_KEY",
|
||
"XAI_API_TOKEN",
|
||
"XAI_ACCESS_TOKEN",
|
||
"GROK_API_KEY",
|
||
"GROK_CODE_XAI_API_KEY",
|
||
"GROK_AUTH",
|
||
"GROK_DEPLOYMENT_KEY",
|
||
"GROK_EXTRA_AUTH_KEY",
|
||
}
|
||
)
|
||
UNMANAGED_MODEL_ROUTE_ENV_VARS = frozenset(
|
||
{
|
||
"GROK_MODELS_BASE_URL",
|
||
"GROK_MODELS_LIST_URL",
|
||
"GROK_XAI_API_BASE_URL",
|
||
"GROK_CLI_CHAT_PROXY_BASE_URL",
|
||
"GROK_IMAGE_GEN_MODEL_OVERRIDE",
|
||
}
|
||
)
|
||
USER_AGENT = "ZhenYangTang-RPA-Grok-Bridge/1.0"
|
||
MANAGED_CONFIG_BEGIN = "# >>> 企业微信 RPA 自动配置（请勿手工修改此区块）"
|
||
MANAGED_CONFIG_END = "# <<< 企业微信 RPA 自动配置结束"
|
||
MANAGED_MCP_PREFIX = "wecom-rpa-"
|
||
CUSTOMER_SERVICE_MCP_NAME = f"{MANAGED_MCP_PREFIX}customer-service"
|
||
VERSION_PATTERN = re.compile(
|
||
r"\d+\.\d+\.\d+(?:-[0-9A-Za-z]+(?:[.-][0-9A-Za-z]+)*)?\Z"
|
||
)
|
||
|
||
|
||
class GrokBuildError(RuntimeError):
|
||
"""Raised for an actionable Grok Build integration failure."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ModelProfile:
|
||
compatible: bool
|
||
profile: str
|
||
model: str
|
||
base_url: str
|
||
api_backend: str
|
||
auth_scheme: str
|
||
temperature: float
|
||
max_completion_tokens: int
|
||
context_window: int
|
||
reason: str
|
||
source_backend: str = ""
|
||
source_base_url: str = ""
|
||
adapter_instance_id: str = ""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ModelSyncResult:
|
||
compatible: bool
|
||
configured: bool
|
||
profile: str
|
||
model: str
|
||
base_url: str
|
||
api_backend: str
|
||
config_path: str
|
||
synced_at: str
|
||
message: str
|
||
source_base_url: str = ""
|
||
source_api_backend: str = ""
|
||
effective_base_url: str = ""
|
||
effective_api_backend: str = ""
|
||
adapter_instance_id: str = ""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ModelEndpointProbe:
|
||
ok: bool
|
||
checked: bool
|
||
api_backend: str
|
||
endpoint: str
|
||
http_status: int | None
|
||
latency_ms: int
|
||
message: str
|
||
detected_protocol: str = ""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class RuntimeStatus:
|
||
installed: bool
|
||
binary_path: str
|
||
version: str
|
||
authenticated: bool
|
||
runtime_home: str
|
||
model_configured: bool
|
||
model_compatible: bool
|
||
model_name: str
|
||
model_message: str
|
||
model_api_backend: str = ""
|
||
model_effective_base_url: str = ""
|
||
adapter_live: bool = False
|
||
warnings: tuple[str, ...] = ()
|
||
|
||
|
||
def _toml_string(value: object) -> str:
|
||
return json.dumps(str(value), ensure_ascii=False)
|
||
|
||
|
||
def _utc_timestamp() -> str:
|
||
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
||
|
||
|
||
def _default_runtime_home(project_dir: Path) -> Path:
|
||
project_key = hashlib.sha256(
|
||
os.path.normcase(str(project_dir.resolve())).encode("utf-8")
|
||
).hexdigest()[:16]
|
||
project_name = re.sub(r"[^A-Za-z0-9_-]+", "-", project_dir.name).strip("-")
|
||
project_name = project_name or "project"
|
||
if os.name == "nt":
|
||
data_root = Path(
|
||
os.environ.get("LOCALAPPDATA")
|
||
or (Path.home() / "AppData" / "Local")
|
||
)
|
||
else:
|
||
data_root = Path(
|
||
os.environ.get("XDG_DATA_HOME")
|
||
or (Path.home() / ".local" / "share")
|
||
)
|
||
return (
|
||
data_root
|
||
/ "ZhenYangTang"
|
||
/ "WeChatRPA"
|
||
/ "GrokBuild"
|
||
/ f"{project_name}-{project_key}"
|
||
/ "custom-agent-only-v1"
|
||
)
|
||
|
||
|
||
DEFAULT_RUNTIME_HOME = _default_runtime_home(PROJECT_DIR)
|
||
_RUNTIME_LOCKS_GUARD = threading.Lock()
|
||
_RUNTIME_LOCKS: dict[str, threading.RLock] = {}
|
||
_MODEL_PROBE_STATE_GUARD = threading.Lock()
|
||
_MODEL_PROBE_LOCKS: dict[str, threading.Lock] = {}
|
||
_MODEL_PROBE_CACHE: dict[
|
||
str,
|
||
tuple[str, float, ModelEndpointProbe],
|
||
] = {}
|
||
|
||
|
||
def _runtime_scope_key(runtime_home: Path) -> str:
|
||
return os.path.normcase(str(runtime_home.resolve()))
|
||
|
||
|
||
def _shared_runtime_lock(runtime_home: Path) -> threading.RLock:
|
||
key = _runtime_scope_key(runtime_home)
|
||
with _RUNTIME_LOCKS_GUARD:
|
||
lock = _RUNTIME_LOCKS.get(key)
|
||
if lock is None:
|
||
lock = threading.RLock()
|
||
_RUNTIME_LOCKS[key] = lock
|
||
return lock
|
||
|
||
|
||
def _shared_model_probe_lock(runtime_home: Path) -> threading.Lock:
|
||
key = _runtime_scope_key(runtime_home)
|
||
with _MODEL_PROBE_STATE_GUARD:
|
||
lock = _MODEL_PROBE_LOCKS.get(key)
|
||
if lock is None:
|
||
lock = threading.Lock()
|
||
_MODEL_PROBE_LOCKS[key] = lock
|
||
return lock
|
||
|
||
|
||
def parse_streaming_event(line: str) -> tuple[str, str]:
|
||
"""Project one Grok streaming-json line into a display category and text."""
|
||
stripped = line.strip()
|
||
if not stripped:
|
||
return "empty", ""
|
||
try:
|
||
event = json.loads(stripped)
|
||
except json.JSONDecodeError:
|
||
return "raw", stripped
|
||
if not isinstance(event, dict):
|
||
return "raw", stripped
|
||
event_type = str(event.get("type") or "event")
|
||
if event_type in {"text", "thought"}:
|
||
return event_type, str(event.get("data") or "")
|
||
if event_type == "error":
|
||
return "error", str(event.get("message") or event.get("data") or "Grok 执行失败")
|
||
if event_type == "end":
|
||
session_id = str(event.get("sessionId") or "")
|
||
stop_reason = str(event.get("stopReason") or "EndTurn")
|
||
turns = event.get("num_turns")
|
||
parts = [f"完成：{stop_reason}"]
|
||
if turns is not None:
|
||
parts.append(f"{turns} 轮")
|
||
if session_id:
|
||
parts.append(f"会话 {session_id}")
|
||
return "end", " · ".join(parts)
|
||
return event_type, stripped
|
||
|
||
|
||
class GrokBuildManager:
|
||
"""Manage an isolated Grok Build sidecar for this project."""
|
||
|
||
def __init__(
|
||
self,
|
||
project_dir: str | os.PathLike[str] = PROJECT_DIR,
|
||
runtime_home: str | os.PathLike[str] | None = None,
|
||
ai_settings_file: str | os.PathLike[str] | None = None,
|
||
integration_settings_file: str | os.PathLike[str] | None = None,
|
||
):
|
||
self.project_dir = Path(project_dir).resolve()
|
||
self.asset_home = (self.project_dir / ".grok-build").resolve()
|
||
self._uses_default_runtime_home = runtime_home is None
|
||
self.runtime_home = Path(
|
||
runtime_home or _default_runtime_home(self.project_dir)
|
||
).resolve()
|
||
self._uses_default_ai_settings = ai_settings_file is None
|
||
self.ai_settings_file = Path(
|
||
ai_settings_file or (self.project_dir / "ai_settings.local.json")
|
||
).resolve()
|
||
if integration_settings_file is None:
|
||
self.integration_defaults_file: Path | None = (
|
||
self.project_dir / "grok_build_settings.json"
|
||
).resolve()
|
||
self.integration_settings_file = (
|
||
self.runtime_home / "integration_settings.json"
|
||
)
|
||
else:
|
||
self.integration_defaults_file = None
|
||
self.integration_settings_file = Path(integration_settings_file).resolve()
|
||
self.user_home = Path.home()
|
||
self.binary_store_home = (
|
||
self.asset_home if self._uses_default_runtime_home else self.runtime_home
|
||
)
|
||
self.bin_dir = self.binary_store_home / "bin"
|
||
self.binary_path = self.bin_dir / ("grok.exe" if os.name == "nt" else "grok")
|
||
self.agent_alias_path = self.bin_dir / ("agent.exe" if os.name == "nt" else "agent")
|
||
self.user_config_file = self.runtime_home / "config.toml"
|
||
# Custom models belong in Grok's normal user config. The upstream
|
||
# `managed_config.toml` filename is reserved for signed enterprise
|
||
# policy and can be refreshed or removed by the runtime.
|
||
self.managed_config_file = self.user_config_file
|
||
self.legacy_managed_config_file = self.runtime_home / "managed_config.toml"
|
||
self.sync_state_file = self.runtime_home / "model_sync.json"
|
||
self.install_state_file = self.binary_store_home / "install.json"
|
||
self._runtime_sync_lock = _shared_runtime_lock(self.runtime_home)
|
||
self._validated_binary_fingerprint: tuple[str, int, int] | None = None
|
||
self._model_probe_scope = _runtime_scope_key(self.runtime_home)
|
||
self._model_probe_lock = _shared_model_probe_lock(self.runtime_home)
|
||
self.migration_warnings: list[str] = []
|
||
if self._uses_default_runtime_home:
|
||
self._migrate_legacy_runtime_state()
|
||
|
||
def _migrate_legacy_runtime_state(self) -> None:
|
||
"""Ignore workspace-era state instead of importing old model sessions.
|
||
|
||
The strict Agent runtime must start without xAI auth or sessions that
|
||
may remember a Grok model. The old files are deliberately left in
|
||
place for manual recovery, but are never placed under ``GROK_HOME``.
|
||
"""
|
||
legacy = self.asset_home
|
||
if not legacy.is_dir() or legacy == self.runtime_home:
|
||
return
|
||
excluded = {
|
||
"bin",
|
||
"downloads",
|
||
"install.json",
|
||
"install.stderr.log",
|
||
"install.stdout.log",
|
||
"grok-page.png",
|
||
"marketplace-cache",
|
||
}
|
||
candidates = [item for item in legacy.iterdir() if item.name not in excluded]
|
||
if not candidates:
|
||
return
|
||
self.migration_warnings.append(
|
||
"检测到旧版 Grok 运行状态，已隔离且不会加载："
|
||
+ "、".join(str(item) for item in candidates)
|
||
)
|
||
|
||
def load_integration_settings(self) -> dict:
|
||
defaults = {
|
||
"binary_path": "",
|
||
"runtime_channel": "stable",
|
||
"default_workspace": str(self.project_dir),
|
||
"default_model": MODEL_PROFILE,
|
||
"context_window": 128000,
|
||
"sync_backend_model": True,
|
||
"sync_mcp_servers": False,
|
||
"customer_service_tools": True,
|
||
"external_compatibility": False,
|
||
"successful_probe_cache_ttl_sec": 300,
|
||
"chat_auto_approve": False,
|
||
}
|
||
sources = [self.integration_defaults_file, self.integration_settings_file]
|
||
for source in sources:
|
||
if source is None:
|
||
continue
|
||
try:
|
||
value = json.loads(source.read_text(encoding="utf-8"))
|
||
if isinstance(value, dict):
|
||
defaults.update(value)
|
||
except (OSError, ValueError, TypeError):
|
||
continue
|
||
return defaults
|
||
|
||
def save_integration_settings(self, values: Mapping[str, object]) -> dict:
|
||
settings = self.load_integration_settings()
|
||
allowed = {
|
||
"binary_path",
|
||
"runtime_channel",
|
||
"default_workspace",
|
||
"default_model",
|
||
"context_window",
|
||
"sync_backend_model",
|
||
"sync_mcp_servers",
|
||
"customer_service_tools",
|
||
"external_compatibility",
|
||
"successful_probe_cache_ttl_sec",
|
||
"chat_auto_approve",
|
||
}
|
||
settings.update({key: values[key] for key in allowed if key in values})
|
||
self.integration_settings_file.parent.mkdir(parents=True, exist_ok=True)
|
||
self._atomic_write(
|
||
self.integration_settings_file,
|
||
json.dumps(settings, ensure_ascii=False, indent=2) + "\n",
|
||
)
|
||
return settings
|
||
|
||
def load_ai_settings(self) -> dict:
|
||
source = self.ai_settings_file
|
||
if (
|
||
self._uses_default_ai_settings
|
||
and not source.is_file()
|
||
and (self.project_dir / "ai_settings.json").is_file()
|
||
):
|
||
source = self.project_dir / "ai_settings.json"
|
||
try:
|
||
value = json.loads(source.read_text(encoding="utf-8"))
|
||
except FileNotFoundError as exc:
|
||
raise GrokBuildError(f"找不到模型配置：{source}") from exc
|
||
except (OSError, ValueError, TypeError) as exc:
|
||
raise GrokBuildError(f"无法读取模型配置：{exc}") from exc
|
||
if not isinstance(value, dict):
|
||
raise GrokBuildError("AI 设置文件根节点必须是 JSON 对象")
|
||
return value
|
||
|
||
@staticmethod
|
||
def _atomic_write(path: Path, content: str) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
temporary = path.with_name(
|
||
f"{path.name}.{os.getpid()}.{threading.get_ident()}."
|
||
f"{uuid.uuid4().hex}.tmp"
|
||
)
|
||
try:
|
||
temporary.write_text(content, encoding="utf-8")
|
||
os.replace(temporary, path)
|
||
finally:
|
||
temporary.unlink(missing_ok=True)
|
||
|
||
@staticmethod
|
||
def _normalize_model_endpoint(api_base: str) -> tuple[bool, str, str, str]:
|
||
value = api_base.strip().rstrip("/")
|
||
if not value:
|
||
return False, "", "", "后台未配置 API 地址"
|
||
try:
|
||
parsed = urlsplit(value)
|
||
except ValueError:
|
||
return False, "", "", "API 地址格式无效"
|
||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||
return False, "", "", "API 地址必须是 http 或 https URL"
|
||
if parsed.username or parsed.password:
|
||
return False, "", "", "API 地址不能包含用户名或密码，请使用 API Key 字段"
|
||
hostname = (parsed.hostname or "").lower()
|
||
if (
|
||
hostname == "x.ai"
|
||
or hostname.endswith(".x.ai")
|
||
or hostname == "grok.com"
|
||
or hostname.endswith(".grok.com")
|
||
):
|
||
return (
|
||
False,
|
||
"",
|
||
"",
|
||
"Grok Agent 必须使用后台自有模型，不能配置 xAI/Grok 模型端点",
|
||
)
|
||
if parsed.query or parsed.fragment:
|
||
return (
|
||
False,
|
||
"",
|
||
"",
|
||
"当前 Grok Build 桥接不支持带 query 或 fragment 的模型地址",
|
||
)
|
||
|
||
path = parsed.path.rstrip("/")
|
||
lowered = path.lower()
|
||
if lowered.endswith("/chat-messages"):
|
||
path = path[: -len("/chat-messages")].rstrip("/")
|
||
normalized = urlunsplit(
|
||
(parsed.scheme, parsed.netloc, path, parsed.query, "")
|
||
).rstrip("/")
|
||
return True, normalized, "dify", "检测到 Dify /chat-messages 协议"
|
||
|
||
endpoints = (
|
||
("/chat/completions", "chat_completions"),
|
||
("/responses", "responses"),
|
||
("/messages", "messages"),
|
||
)
|
||
for suffix, backend in endpoints:
|
||
if lowered.endswith(suffix):
|
||
path = path[: -len(suffix)].rstrip("/")
|
||
normalized = urlunsplit(
|
||
(parsed.scheme, parsed.netloc, path, parsed.query, "")
|
||
).rstrip("/")
|
||
return True, normalized, backend, "协议已从完整端点自动识别"
|
||
|
||
normalized = urlunsplit(
|
||
(parsed.scheme, parsed.netloc, path, parsed.query, "")
|
||
).rstrip("/")
|
||
return (
|
||
True,
|
||
normalized,
|
||
"chat_completions",
|
||
"按 OpenAI Chat Completions 兼容地址配置",
|
||
)
|
||
|
||
def model_profile(self, ai_settings: Mapping[str, object] | None = None) -> ModelProfile:
|
||
settings = dict(ai_settings or self.load_ai_settings())
|
||
dedicated = bool(settings.get("GROK_MODEL_ENABLED", False))
|
||
api_base_key = "GROK_API_BASE" if dedicated else "AI_API_BASE"
|
||
model_key = "GROK_MODEL" if dedicated else "AI_MODEL"
|
||
temperature_key = "GROK_TEMPERATURE" if dedicated else "AI_TEMPERATURE"
|
||
max_tokens_key = "GROK_MAX_TOKENS" if dedicated else "AI_MAX_TOKENS"
|
||
compatible, base_url, backend, reason = self._normalize_model_endpoint(
|
||
str(settings.get(api_base_key) or "")
|
||
)
|
||
if dedicated:
|
||
configured_backend = str(
|
||
settings.get("GROK_API_BACKEND") or "chat_completions"
|
||
).strip()
|
||
if configured_backend in {
|
||
"chat_completions",
|
||
"responses",
|
||
"messages",
|
||
"dify",
|
||
}:
|
||
if compatible:
|
||
if backend == "dify" and configured_backend != "dify":
|
||
compatible = False
|
||
reason = (
|
||
"API 地址是 Dify /chat-messages；请把接口协议选择为 "
|
||
"Dify Chat Messages（本地工具调用适配）"
|
||
)
|
||
else:
|
||
backend = configured_backend
|
||
if backend == "dify":
|
||
reason = (
|
||
"Dify Chat Messages 将通过项目内置本地工具调用适配器"
|
||
"接入 Grok Build"
|
||
)
|
||
else:
|
||
compatible = False
|
||
reason = f"后台配置了不支持的 Agent 自有模型协议：{configured_backend}"
|
||
model = str(settings.get(model_key) or "").strip()
|
||
if compatible and backend == "dify" and not model:
|
||
model = "dify-app"
|
||
if compatible and not model:
|
||
compatible = False
|
||
reason = "后台未配置 Agent 自有模型名称" if dedicated else "后台未配置模型名称"
|
||
key_name = "GROK_API_KEY" if dedicated else "AI_API_KEY"
|
||
if compatible and not str(settings.get(key_name) or "").strip():
|
||
compatible = False
|
||
reason = (
|
||
"后台未配置 Agent 自有模型 API Key；"
|
||
"自定义模型必须提供独立密钥，系统不会使用 xAI 登录凭据"
|
||
)
|
||
configured_auth = str(
|
||
settings.get("GROK_AUTH_SCHEME") or "auto"
|
||
).strip().lower() if dedicated else "auto"
|
||
if configured_auth not in {"auto", "bearer", "x_api_key"}:
|
||
compatible = False
|
||
reason = f"后台配置了不支持的认证方式：{configured_auth}"
|
||
configured_auth = "bearer"
|
||
if configured_auth == "auto":
|
||
hostname = (urlsplit(base_url).hostname or "").lower() if base_url else ""
|
||
auth_scheme = (
|
||
"x_api_key"
|
||
if backend == "messages"
|
||
and (hostname == "api.anthropic.com" or hostname.endswith(".anthropic.com"))
|
||
else "bearer"
|
||
)
|
||
else:
|
||
auth_scheme = configured_auth
|
||
if backend == "dify":
|
||
if configured_auth == "x_api_key":
|
||
compatible = False
|
||
reason = "Dify Chat Messages 必须使用 Authorization Bearer 认证"
|
||
auth_scheme = "bearer"
|
||
try:
|
||
temperature = min(
|
||
2.0, max(0.0, float(settings.get(temperature_key, 0.7)))
|
||
)
|
||
except (TypeError, ValueError):
|
||
temperature = 0.7
|
||
try:
|
||
max_tokens = min(
|
||
262144, max(64, int(settings.get(max_tokens_key, 8192)))
|
||
)
|
||
except (TypeError, ValueError):
|
||
max_tokens = 8192
|
||
integration = self.load_integration_settings()
|
||
try:
|
||
context_window = min(
|
||
2_000_000,
|
||
max(
|
||
4096,
|
||
int(
|
||
settings.get("GROK_CONTEXT_WINDOW", 128000)
|
||
if dedicated
|
||
else integration.get("context_window", 128000)
|
||
),
|
||
),
|
||
)
|
||
except (TypeError, ValueError):
|
||
context_window = 128000
|
||
return ModelProfile(
|
||
compatible=compatible,
|
||
profile=MODEL_PROFILE,
|
||
model=model,
|
||
base_url=base_url,
|
||
api_backend=backend,
|
||
auth_scheme=auth_scheme,
|
||
temperature=temperature,
|
||
max_completion_tokens=max_tokens,
|
||
context_window=context_window,
|
||
reason=reason,
|
||
)
|
||
|
||
def _agent_uses_dify_source(self) -> bool:
|
||
"""Return whether the managed Agent model is configured through Dify.
|
||
|
||
Grok Build's built-in ``web_search`` helper is not an ordinary function
|
||
tool: it always calls a model endpoint that implements the Responses API
|
||
and native web search. The loopback Dify adapter intentionally exposes
|
||
Chat Completions only, so advertising that helper would make every web
|
||
search call ``/v1/responses`` and fail with HTTP 404.
|
||
"""
|
||
try:
|
||
settings = self.load_ai_settings()
|
||
except GrokBuildError:
|
||
return False
|
||
return bool(
|
||
settings.get("GROK_MODEL_ENABLED", False)
|
||
and str(settings.get("GROK_API_BACKEND") or "").strip().lower()
|
||
== "dify"
|
||
)
|
||
|
||
def agent_model_profile(
|
||
self,
|
||
ai_settings: Mapping[str, object] | None = None,
|
||
) -> ModelProfile:
|
||
"""Return the dedicated model allowed to power Grok Build Agent.
|
||
|
||
The normal customer-service ``AI_*`` endpoint is intentionally not a
|
||
fallback. A configured Dify application is exposed to Grok through
|
||
the loopback-only protocol adapter, while native model endpoints pass
|
||
through unchanged.
|
||
"""
|
||
settings = dict(ai_settings or self.load_ai_settings())
|
||
if bool(settings.get("GROK_MODEL_ENABLED", False)):
|
||
profile = self.model_profile(settings)
|
||
if not profile.compatible or profile.api_backend != "dify":
|
||
return profile
|
||
try:
|
||
from dify_grok_adapter import ensure_dify_adapter
|
||
|
||
try:
|
||
adapter_timeout = int(
|
||
settings.get("GROK_CUSTOMER_SERVICE_TIMEOUT", 180)
|
||
)
|
||
except (TypeError, ValueError):
|
||
adapter_timeout = 180
|
||
adapter = ensure_dify_adapter(
|
||
str(self.runtime_home),
|
||
upstream_base_url=profile.base_url,
|
||
api_key=str(settings.get("GROK_API_KEY") or ""),
|
||
model=profile.model,
|
||
timeout=adapter_timeout,
|
||
inputs=(
|
||
settings.get("GROK_DIFY_INPUTS")
|
||
if isinstance(
|
||
settings.get("GROK_DIFY_INPUTS"),
|
||
Mapping,
|
||
)
|
||
else {}
|
||
),
|
||
)
|
||
except Exception as exc:
|
||
return replace(
|
||
profile,
|
||
compatible=False,
|
||
reason=f"无法启动 Dify 本地工具调用适配器：{exc}",
|
||
source_backend="dify",
|
||
source_base_url=profile.base_url,
|
||
)
|
||
return replace(
|
||
profile,
|
||
base_url=adapter.base_url,
|
||
api_backend="chat_completions",
|
||
auth_scheme="bearer",
|
||
reason=(
|
||
"Dify Chat Messages 已通过项目内置本地工具调用适配器接入"
|
||
),
|
||
source_backend="dify",
|
||
source_base_url=profile.base_url,
|
||
adapter_instance_id=adapter.instance_id,
|
||
)
|
||
return ModelProfile(
|
||
compatible=False,
|
||
profile=MODEL_PROFILE,
|
||
model="",
|
||
base_url="",
|
||
api_backend="",
|
||
auth_scheme="bearer",
|
||
temperature=0.3,
|
||
max_completion_tokens=8192,
|
||
context_window=128000,
|
||
reason=(
|
||
"后台尚未启用 Grok Agent 自有模型；请配置 GROK_API_BASE、"
|
||
"GROK_API_KEY 和 GROK_MODEL。Agent 不会回退到 Grok/xAI 模型"
|
||
),
|
||
)
|
||
|
||
def agent_model_api_key(
|
||
self,
|
||
ai_settings: Mapping[str, object] | None = None,
|
||
*,
|
||
profile: ModelProfile | None = None,
|
||
) -> str:
|
||
"""Return the credential Grok may receive for the effective endpoint.
|
||
|
||
Native providers use their configured key. Dify uses a random
|
||
loopback-adapter token so the Dify application key never enters the
|
||
Grok child process.
|
||
"""
|
||
settings = dict(ai_settings or self.load_ai_settings())
|
||
effective = profile or self.agent_model_profile(settings)
|
||
if not effective.compatible:
|
||
raise GrokBuildError(effective.reason)
|
||
if effective.source_backend != "dify":
|
||
key = str(settings.get("GROK_API_KEY") or "").strip()
|
||
if not key:
|
||
raise GrokBuildError("后台 Agent 自有模型缺少独立 API Key")
|
||
return key
|
||
try:
|
||
from dify_grok_adapter import ensure_dify_adapter
|
||
|
||
try:
|
||
adapter_timeout = int(
|
||
settings.get("GROK_CUSTOMER_SERVICE_TIMEOUT", 180)
|
||
)
|
||
except (TypeError, ValueError):
|
||
adapter_timeout = 180
|
||
adapter = ensure_dify_adapter(
|
||
str(self.runtime_home),
|
||
upstream_base_url=effective.source_base_url,
|
||
api_key=str(settings.get("GROK_API_KEY") or ""),
|
||
model=effective.model,
|
||
timeout=adapter_timeout,
|
||
inputs=(
|
||
settings.get("GROK_DIFY_INPUTS")
|
||
if isinstance(
|
||
settings.get("GROK_DIFY_INPUTS"),
|
||
Mapping,
|
||
)
|
||
else {}
|
||
),
|
||
)
|
||
except Exception as exc:
|
||
raise GrokBuildError(
|
||
f"无法取得 Dify 本地适配器凭据：{exc}"
|
||
) from exc
|
||
if adapter.base_url != effective.base_url:
|
||
raise GrokBuildError(
|
||
"Dify 本地适配器端口在配置核验期间发生变化，请重新同步"
|
||
)
|
||
return adapter.local_api_key
|
||
|
||
@staticmethod
|
||
def _model_operation_endpoint(profile: ModelProfile) -> str:
|
||
suffixes = {
|
||
"chat_completions": "chat/completions",
|
||
"responses": "responses",
|
||
"messages": "messages",
|
||
}
|
||
suffix = suffixes.get(profile.api_backend, "")
|
||
if not suffix or not profile.base_url:
|
||
return ""
|
||
return f"{profile.base_url.rstrip('/')}/{suffix}"
|
||
|
||
@staticmethod
|
||
def _model_probe_payload(profile: ModelProfile) -> dict[str, object]:
|
||
if profile.source_backend == "dify":
|
||
challenge = uuid.uuid4().hex
|
||
return {
|
||
"model": profile.model,
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": (
|
||
"调用 health_check 工具完成协议预检，token 必须为 "
|
||
f"{challenge}。"
|
||
),
|
||
}
|
||
],
|
||
"tools": [
|
||
{
|
||
"type": "function",
|
||
"function": {
|
||
"name": "health_check",
|
||
"description": "完成本地协议预检；不执行外部操作。",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"token": {
|
||
"type": "string",
|
||
"const": challenge,
|
||
}
|
||
},
|
||
"required": ["token"],
|
||
"additionalProperties": False,
|
||
},
|
||
},
|
||
}
|
||
],
|
||
"tool_choice": {
|
||
"type": "function",
|
||
"function": {"name": "health_check"},
|
||
},
|
||
"max_tokens": 32,
|
||
"stream": True,
|
||
}
|
||
if profile.api_backend == "responses":
|
||
return {
|
||
"model": profile.model,
|
||
"input": "health check",
|
||
"max_output_tokens": 8,
|
||
"stream": True,
|
||
}
|
||
if profile.api_backend == "messages":
|
||
return {
|
||
"model": profile.model,
|
||
"messages": [{"role": "user", "content": "health check"}],
|
||
"max_tokens": 8,
|
||
"stream": True,
|
||
}
|
||
return {
|
||
"model": profile.model,
|
||
"messages": [{"role": "user", "content": "health check"}],
|
||
"max_tokens": 8,
|
||
"stream": True,
|
||
}
|
||
|
||
@staticmethod
|
||
def _unauthenticated_route_status(
|
||
endpoint: str,
|
||
*,
|
||
timeout: float,
|
||
) -> int | None:
|
||
"""Return only a route status; never attach a model credential."""
|
||
request = urllib.request.Request(
|
||
endpoint,
|
||
data=b"{}",
|
||
headers={
|
||
"User-Agent": USER_AGENT,
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json",
|
||
},
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=timeout) as response:
|
||
return int(response.getcode())
|
||
except urllib.error.HTTPError as exc:
|
||
try:
|
||
return int(exc.code)
|
||
finally:
|
||
exc.close()
|
||
except (OSError, urllib.error.URLError, ValueError):
|
||
return None
|
||
|
||
def _detect_endpoint_protocol(
|
||
self,
|
||
profile: ModelProfile,
|
||
*,
|
||
timeout: float,
|
||
) -> tuple[str, str]:
|
||
"""Detect a nearby route after a 404 without sending the API key."""
|
||
base = profile.base_url.rstrip("/")
|
||
route_timeout = min(3.0, max(1.0, timeout / 3.0))
|
||
labels = {
|
||
"chat_completions": "OpenAI Chat Completions",
|
||
"responses": "OpenAI Responses",
|
||
"messages": "Anthropic Messages",
|
||
}
|
||
suffixes = {
|
||
"chat_completions": "chat/completions",
|
||
"responses": "responses",
|
||
"messages": "messages",
|
||
}
|
||
recognized_statuses = {200, 201, 400, 401, 403, 422, 429}
|
||
for backend in ("responses", "chat_completions", "messages"):
|
||
if backend == profile.api_backend:
|
||
continue
|
||
endpoint = f"{base}/{suffixes[backend]}"
|
||
status = self._unauthenticated_route_status(
|
||
endpoint,
|
||
timeout=route_timeout,
|
||
)
|
||
if status in recognized_statuses:
|
||
return backend, labels[backend]
|
||
|
||
dify_endpoint = f"{base}/chat-messages"
|
||
dify_status = self._unauthenticated_route_status(
|
||
dify_endpoint,
|
||
timeout=route_timeout,
|
||
)
|
||
if dify_status in recognized_statuses:
|
||
return "dify_chat_messages", "Dify /chat-messages"
|
||
return "", ""
|
||
|
||
@staticmethod
|
||
def _probe_failure_message(
|
||
profile: ModelProfile,
|
||
endpoint: str,
|
||
status: int,
|
||
detected_protocol: str,
|
||
detected_label: str,
|
||
) -> str:
|
||
protocol_labels = {
|
||
"chat_completions": "OpenAI Chat Completions",
|
||
"responses": "OpenAI Responses",
|
||
"messages": "Anthropic Messages",
|
||
}
|
||
selected = (
|
||
"Dify Chat Messages 本地适配器"
|
||
if profile.source_backend == "dify"
|
||
else protocol_labels.get(profile.api_backend, profile.api_backend)
|
||
)
|
||
if status == 404 and detected_protocol == "dify_chat_messages":
|
||
return (
|
||
f"{selected} 端点返回 HTTP 404：{endpoint}。检测到同一服务实际"
|
||
"提供 Dify /chat-messages；请在管理后台把接口协议改为 "
|
||
"Dify Chat Messages（本地工具调用适配）。桌面端会自动创建"
|
||
"本机适配端点并验证工具调用协议。"
|
||
)
|
||
if status == 404 and detected_protocol:
|
||
return (
|
||
f"{selected} 端点返回 HTTP 404：{endpoint}。同一基础地址检测到"
|
||
f" {detected_label}，请在管理后台选择匹配的接口协议。"
|
||
)
|
||
if status == 404:
|
||
return (
|
||
f"{selected} 端点或模型不存在（HTTP 404）：{endpoint}。"
|
||
"请核对 API 基址、接口协议和模型名称。"
|
||
)
|
||
if status in {401, 403}:
|
||
return (
|
||
f"{selected} 端点认证失败（HTTP {status}）。请核对自有模型 "
|
||
"API Key 和认证方式。"
|
||
)
|
||
if status == 429:
|
||
return (
|
||
f"{selected} 端点当前限流或额度不足（HTTP 429），暂不能启动 "
|
||
"Agent。"
|
||
)
|
||
if status in {400, 405, 415, 422}:
|
||
return (
|
||
f"{selected} 端点拒绝了 Grok Build 兼容请求（HTTP {status}）。"
|
||
"请确认服务实现了所选协议及流式生成。"
|
||
)
|
||
return f"{selected} 端点预检失败（HTTP {status}）：{endpoint}"
|
||
|
||
def probe_agent_model(
|
||
self,
|
||
ai_settings: Mapping[str, object] | None = None,
|
||
*,
|
||
timeout: float = 12.0,
|
||
force: bool = False,
|
||
cache_ttl: float = 30.0,
|
||
) -> ModelEndpointProbe:
|
||
"""Verify the exact model route Grok Build will call.
|
||
|
||
This performs a tiny streaming request with the configured custom-model
|
||
credential. The result and all errors are deliberately secret-free.
|
||
"""
|
||
settings = dict(ai_settings or self.load_ai_settings())
|
||
profile = self.agent_model_profile(settings)
|
||
endpoint = self._model_operation_endpoint(profile)
|
||
reported_backend = profile.source_backend or profile.api_backend
|
||
if not profile.compatible or not endpoint:
|
||
return ModelEndpointProbe(
|
||
ok=False,
|
||
checked=False,
|
||
api_backend=reported_backend,
|
||
endpoint=endpoint,
|
||
http_status=None,
|
||
latency_ms=0,
|
||
message=profile.reason or "后台自有模型配置不完整",
|
||
)
|
||
|
||
upstream_api_key = str(settings.get("GROK_API_KEY") or "")
|
||
api_key = self.agent_model_api_key(settings, profile=profile)
|
||
key_digest = hashlib.sha256(
|
||
upstream_api_key.encode("utf-8")
|
||
).hexdigest()
|
||
cache_key = hashlib.sha256(
|
||
json.dumps(
|
||
{
|
||
"base_url": profile.base_url,
|
||
"api_backend": profile.api_backend,
|
||
"source_base_url": profile.source_base_url,
|
||
"source_backend": profile.source_backend,
|
||
"adapter_instance_id": profile.adapter_instance_id,
|
||
"auth_scheme": profile.auth_scheme,
|
||
"model": profile.model,
|
||
"key_digest": key_digest,
|
||
},
|
||
ensure_ascii=False,
|
||
sort_keys=True,
|
||
).encode("utf-8")
|
||
).hexdigest()
|
||
|
||
with self._model_probe_lock:
|
||
now = time.monotonic()
|
||
with _MODEL_PROBE_STATE_GUARD:
|
||
cached = _MODEL_PROBE_CACHE.get(self._model_probe_scope)
|
||
requested_ttl = max(0.0, float(cache_ttl))
|
||
effective_ttl = requested_ttl
|
||
if cached is not None and cached[2].ok and requested_ttl > 0:
|
||
integration = self.load_integration_settings()
|
||
try:
|
||
successful_ttl = float(
|
||
integration.get(
|
||
"successful_probe_cache_ttl_sec",
|
||
300,
|
||
)
|
||
)
|
||
except (TypeError, ValueError):
|
||
successful_ttl = 300.0
|
||
effective_ttl = max(
|
||
requested_ttl,
|
||
min(3600.0, max(300.0, successful_ttl)),
|
||
)
|
||
if (
|
||
not force
|
||
and cached is not None
|
||
and cached[0] == cache_key
|
||
and now - cached[1] <= effective_ttl
|
||
):
|
||
return cached[2]
|
||
|
||
headers = {
|
||
"User-Agent": USER_AGENT,
|
||
"Content-Type": "application/json",
|
||
"Accept": "text/event-stream, application/json",
|
||
}
|
||
if profile.auth_scheme == "x_api_key":
|
||
headers["x-api-key"] = api_key
|
||
headers["anthropic-version"] = "2023-06-01"
|
||
else:
|
||
headers["Authorization"] = f"Bearer {api_key}"
|
||
request = urllib.request.Request(
|
||
endpoint,
|
||
data=json.dumps(
|
||
self._model_probe_payload(profile),
|
||
ensure_ascii=False,
|
||
).encode("utf-8"),
|
||
headers=headers,
|
||
method="POST",
|
||
)
|
||
started = time.monotonic()
|
||
try:
|
||
with urllib.request.urlopen(
|
||
request,
|
||
timeout=max(1.0, float(timeout)),
|
||
) as response:
|
||
status = int(response.getcode())
|
||
content_type = str(
|
||
response.headers.get("Content-Type", "")
|
||
).lower()
|
||
latency_ms = max(0, int((time.monotonic() - started) * 1000))
|
||
if 200 <= status < 300 and "text/event-stream" in content_type:
|
||
insecure_dify = bool(
|
||
profile.source_backend == "dify"
|
||
and profile.source_base_url.lower().startswith("http://")
|
||
and not re.match(
|
||
r"^http://(?:127\.0\.0\.1|localhost|\[::1\])(?::|/|$)",
|
||
profile.source_base_url,
|
||
flags=re.I,
|
||
)
|
||
)
|
||
result = ModelEndpointProbe(
|
||
ok=True,
|
||
checked=True,
|
||
api_backend=reported_backend,
|
||
endpoint=endpoint,
|
||
http_status=status,
|
||
latency_ms=latency_ms,
|
||
message=(
|
||
(
|
||
"警告：Dify 上游使用明文 HTTP，API Key 与会话"
|
||
"内容未加密传输。请尽快改用 HTTPS。"
|
||
if insecure_dify
|
||
else ""
|
||
)
|
||
+ (
|
||
"Dify 本地工具调用适配器预检通过"
|
||
if profile.source_backend == "dify"
|
||
else "模型端点预检通过"
|
||
)
|
||
+ f"（HTTP {status}，{latency_ms}ms）："
|
||
+ (
|
||
profile.source_base_url
|
||
if profile.source_backend == "dify"
|
||
else endpoint
|
||
)
|
||
),
|
||
)
|
||
elif 200 <= status < 300:
|
||
result = ModelEndpointProbe(
|
||
ok=False,
|
||
checked=True,
|
||
api_backend=reported_backend,
|
||
endpoint=endpoint,
|
||
http_status=status,
|
||
latency_ms=latency_ms,
|
||
message=(
|
||
f"模型端点返回 HTTP {status}，但没有提供 Grok Build "
|
||
"所需的 text/event-stream 流式响应。请核对接口协议。"
|
||
),
|
||
)
|
||
else:
|
||
result = ModelEndpointProbe(
|
||
ok=False,
|
||
checked=True,
|
||
api_backend=reported_backend,
|
||
endpoint=endpoint,
|
||
http_status=status,
|
||
latency_ms=latency_ms,
|
||
message=self._probe_failure_message(
|
||
profile,
|
||
endpoint,
|
||
status,
|
||
"",
|
||
"",
|
||
),
|
||
)
|
||
except urllib.error.HTTPError as exc:
|
||
status = int(exc.code)
|
||
exc.close()
|
||
latency_ms = max(0, int((time.monotonic() - started) * 1000))
|
||
detected_protocol = ""
|
||
detected_label = ""
|
||
if status == 404:
|
||
detected_protocol, detected_label = (
|
||
self._detect_endpoint_protocol(
|
||
profile,
|
||
timeout=max(1.0, float(timeout)),
|
||
)
|
||
)
|
||
result = ModelEndpointProbe(
|
||
ok=False,
|
||
checked=True,
|
||
api_backend=reported_backend,
|
||
endpoint=endpoint,
|
||
http_status=status,
|
||
latency_ms=latency_ms,
|
||
message=self._probe_failure_message(
|
||
profile,
|
||
endpoint,
|
||
status,
|
||
detected_protocol,
|
||
detected_label,
|
||
),
|
||
detected_protocol=detected_protocol,
|
||
)
|
||
except (OSError, urllib.error.URLError, ValueError) as exc:
|
||
latency_ms = max(0, int((time.monotonic() - started) * 1000))
|
||
reason = str(getattr(exc, "reason", exc) or "连接失败")
|
||
result = ModelEndpointProbe(
|
||
ok=False,
|
||
checked=True,
|
||
api_backend=reported_backend,
|
||
endpoint=endpoint,
|
||
http_status=None,
|
||
latency_ms=latency_ms,
|
||
message=(
|
||
f"无法连接后台自有模型端点：{endpoint}（{reason}）。"
|
||
"请检查服务器、网络和端口。"
|
||
),
|
||
)
|
||
|
||
with _MODEL_PROBE_STATE_GUARD:
|
||
_MODEL_PROBE_CACHE[self._model_probe_scope] = (
|
||
cache_key,
|
||
time.monotonic(),
|
||
result,
|
||
)
|
||
return result
|
||
|
||
def _render_managed_config(
|
||
self,
|
||
profile: ModelProfile,
|
||
ai_settings: Mapping[str, object],
|
||
include_mcp: bool,
|
||
include_customer_service_tools: bool = False,
|
||
subagents_enabled: bool = True,
|
||
external_compatibility: bool = False,
|
||
disabled_plugins: Sequence[str] = (),
|
||
disabled_external_mcp_names: Sequence[str] = (),
|
||
) -> str:
|
||
if profile.compatible and profile.api_backend == "dify":
|
||
raise GrokBuildError(
|
||
"Dify 来源协议不能直接写入 Grok 配置，必须先解析本地适配端点"
|
||
)
|
||
if profile.compatible and profile.source_backend == "dify":
|
||
parsed_adapter = urlsplit(profile.base_url)
|
||
if (
|
||
parsed_adapter.hostname not in {"127.0.0.1", "localhost", "::1"}
|
||
or profile.api_backend != "chat_completions"
|
||
):
|
||
raise GrokBuildError(
|
||
"Dify 的 Grok 有效模型必须指向本机 Chat Completions 适配器"
|
||
)
|
||
lines = [
|
||
MANAGED_CONFIG_BEGIN,
|
||
"# 由企业微信 RPA 根据后台配置自动生成。",
|
||
"# API Key 不写入此文件，由 WECOM_GROK_API_KEY 环境变量注入。",
|
||
"",
|
||
"[compat.cursor]",
|
||
*[
|
||
f"{surface} = {'true' if external_compatibility else 'false'}"
|
||
for surface in (
|
||
"skills",
|
||
"rules",
|
||
"agents",
|
||
"mcps",
|
||
"hooks",
|
||
"sessions",
|
||
)
|
||
],
|
||
"",
|
||
"[compat.claude]",
|
||
*[
|
||
f"{surface} = {'true' if external_compatibility else 'false'}"
|
||
for surface in (
|
||
"skills",
|
||
"rules",
|
||
"agents",
|
||
"mcps",
|
||
"hooks",
|
||
"sessions",
|
||
)
|
||
],
|
||
"",
|
||
"[compat.codex]",
|
||
f"sessions = {'true' if external_compatibility else 'false'}",
|
||
"",
|
||
"[plugins]",
|
||
f"disabled = {self._toml_value(sorted(set(disabled_plugins)))}",
|
||
]
|
||
if not external_compatibility:
|
||
# A native permission table prevents Grok from falling back to
|
||
# ~/.claude/settings*.json, whose rules are otherwise loaded even
|
||
# when the regular Claude compatibility cells are disabled.
|
||
lines.extend(["", "[permission]", "rules = []"])
|
||
if profile.compatible:
|
||
lines.extend(
|
||
[
|
||
"",
|
||
"[models]",
|
||
f"default = {_toml_string(profile.profile)}",
|
||
f"allowed_models = [{_toml_string(profile.profile)}]",
|
||
f"web_search = {_toml_string(profile.profile)}",
|
||
f"session_summary = {_toml_string(profile.profile)}",
|
||
f"image_description = {_toml_string(profile.profile)}",
|
||
f"prompt_suggestion = {_toml_string(profile.profile)}",
|
||
"",
|
||
"[ui]",
|
||
"prompt_suggestions = false",
|
||
f"fork_secondary_model = {_toml_string(profile.profile)}",
|
||
"",
|
||
"[suggestions]",
|
||
"enabled = false",
|
||
"ai_enabled = false",
|
||
f"ai_model = {_toml_string(profile.profile)}",
|
||
"",
|
||
"[subagents]",
|
||
f"enabled = {'true' if subagents_enabled else 'false'}",
|
||
"",
|
||
"[subagents.models]",
|
||
*[
|
||
f"{self._toml_key(name)} = {_toml_string(profile.profile)}"
|
||
for name in sorted(PINNED_SUBAGENT_NAMES)
|
||
],
|
||
"",
|
||
"[goal]",
|
||
"use_current_model_only = true",
|
||
"",
|
||
"[auto_mode]",
|
||
f"classifier_model = {_toml_string(profile.profile)}",
|
||
"",
|
||
"[compaction.memory_flush]",
|
||
f"flush_model = {_toml_string(profile.profile)}",
|
||
"",
|
||
f"[model.{profile.profile}]",
|
||
f"model = {_toml_string(profile.model)}",
|
||
f"base_url = {_toml_string(profile.base_url)}",
|
||
f"name = {_toml_string('后台模型 · ' + profile.model)}",
|
||
f"env_key = {_toml_string(MODEL_API_KEY_ENV)}",
|
||
f"api_backend = {_toml_string(profile.api_backend)}",
|
||
f"temperature = {profile.temperature:.6g}",
|
||
f"max_completion_tokens = {profile.max_completion_tokens}",
|
||
f"context_window = {profile.context_window}",
|
||
]
|
||
)
|
||
if profile.source_backend != "dify":
|
||
lines.append(
|
||
f"auth_scheme = {_toml_string(profile.auth_scheme)}"
|
||
)
|
||
if profile.api_backend == "messages" and profile.auth_scheme == "x_api_key":
|
||
lines.extend(
|
||
[
|
||
'extra_headers = { "anthropic-version" = "2023-06-01" }',
|
||
]
|
||
)
|
||
managed_mcp_servers: list[object] = []
|
||
if include_customer_service_tools:
|
||
managed_mcp_servers.extend(self._customer_service_mcp_servers(ai_settings))
|
||
if include_mcp and isinstance(ai_settings.get("AI_MCP_SERVERS"), list):
|
||
managed_mcp_servers.extend(ai_settings["AI_MCP_SERVERS"])
|
||
managed_mcp_names: set[str] = set()
|
||
if managed_mcp_servers:
|
||
managed_mcp_lines = self._render_mcp_servers(managed_mcp_servers)
|
||
lines.extend(managed_mcp_lines)
|
||
managed_mcp_names = self._configured_mcp_names(
|
||
"\n".join(managed_mcp_lines)
|
||
)
|
||
for name in sorted(
|
||
{
|
||
str(value).strip()
|
||
for value in disabled_external_mcp_names
|
||
if str(value).strip()
|
||
}
|
||
- managed_mcp_names
|
||
):
|
||
lines.extend(
|
||
[
|
||
"",
|
||
f"[mcp_servers.{self._toml_key(name)}]",
|
||
"enabled = false",
|
||
]
|
||
)
|
||
lines.extend(["", MANAGED_CONFIG_END, ""])
|
||
return "\n".join(lines)
|
||
|
||
def _customer_service_mcp_servers(
|
||
self,
|
||
ai_settings: Mapping[str, object] | None = None,
|
||
) -> list[dict[str, object]]:
|
||
"""Return the trusted project-local MCP adapter exposed to Grok Build.
|
||
|
||
Credentials are deliberately absent from the TOML. The server exposes
|
||
only deterministic, session-scoped context, intent, validation, and
|
||
pending-registration operations; Grok itself generates the reply and
|
||
the MCP can never send a message to WeCom.
|
||
"""
|
||
settings = dict(ai_settings or {})
|
||
try:
|
||
timeout = int(
|
||
settings.get("GROK_CUSTOMER_SERVICE_TIMEOUT", 180) or 180
|
||
)
|
||
except (TypeError, ValueError):
|
||
timeout = 180
|
||
return [
|
||
{
|
||
"name": "customer-service",
|
||
"transport": "stdio",
|
||
"command": sys.executable,
|
||
"args": [str(self.project_dir / "grok_customer_service_mcp.py")],
|
||
"cwd": str(self.project_dir),
|
||
"startup_timeout_sec": 30,
|
||
"tool_timeout_sec": min(900, max(30, timeout + 30)),
|
||
}
|
||
]
|
||
|
||
@staticmethod
|
||
def _safe_toml_identifier(value: str) -> str:
|
||
cleaned = re.sub(r"[^A-Za-z0-9_-]+", "-", value.strip()).strip("-")
|
||
return cleaned or "server"
|
||
|
||
@staticmethod
|
||
def _mcp_value_env_name(server: str, category: str, key: object) -> str:
|
||
identity = f"{server}\0{category}\0{key}".encode("utf-8")
|
||
digest = hashlib.sha256(identity).hexdigest()[:16].upper()
|
||
return f"WECOM_GROK_MCP_{digest}"
|
||
|
||
def _mcp_config_value(
|
||
self,
|
||
server: str,
|
||
category: str,
|
||
key: object,
|
||
value: object,
|
||
) -> str:
|
||
rendered = str(value)
|
||
if re.fullmatch(r"\$\{[A-Za-z_][A-Za-z0-9_]*\}", rendered):
|
||
variable = rendered[2:-1]
|
||
if (
|
||
variable == MODEL_API_KEY_ENV
|
||
or variable.startswith("WECOM_GROK_MCP_")
|
||
):
|
||
raise GrokBuildError(
|
||
f"{variable} 是桥接器保留的凭据变量，不能手工引用"
|
||
)
|
||
return rendered
|
||
if rendered.startswith("${") and rendered.endswith("}"):
|
||
raise GrokBuildError(
|
||
"MCP 环境变量引用仅支持 ${VAR}，不支持默认值或嵌套表达式"
|
||
)
|
||
env_name = self._mcp_value_env_name(server, category, key)
|
||
return f"${{{env_name}}}"
|
||
|
||
def _render_mcp_servers(self, raw_servers: object) -> list[str]:
|
||
if not isinstance(raw_servers, list):
|
||
return []
|
||
output: list[str] = []
|
||
used: set[str] = set()
|
||
for index, raw in enumerate(raw_servers, start=1):
|
||
if not isinstance(raw, dict) or raw.get("enabled") is False:
|
||
continue
|
||
base_name = self._safe_toml_identifier(
|
||
str(raw.get("name") or raw.get("id") or f"server-{index}")
|
||
)
|
||
base_name = f"{MANAGED_MCP_PREFIX}{base_name}"
|
||
name = base_name
|
||
suffix = 2
|
||
while name in used:
|
||
name = f"{base_name}-{suffix}"
|
||
suffix += 1
|
||
used.add(name)
|
||
transport = str(raw.get("transport") or "").strip().lower()
|
||
command = str(raw.get("command") or "").strip()
|
||
url = str(raw.get("url") or "").strip()
|
||
if not transport:
|
||
transport = "stdio" if command else "http"
|
||
if transport == "stdio" and not command:
|
||
continue
|
||
if transport not in {"stdio", "sse", "http", "streamable_http"}:
|
||
continue
|
||
if transport != "stdio" and not url:
|
||
continue
|
||
secret_option = re.compile(
|
||
r"(?i)(?:^|[\s=])--?(?:api[-_]?key|access[-_]?token|token|"
|
||
r"secret|password|credential)(?:=|\s|$)"
|
||
)
|
||
if transport == "stdio" and secret_option.search(command):
|
||
raise GrokBuildError(
|
||
f"MCP {base_name} 的 command 含疑似密钥参数；请移到 env"
|
||
)
|
||
if transport != "stdio":
|
||
try:
|
||
parsed_url = urlsplit(url)
|
||
except ValueError as exc:
|
||
raise GrokBuildError(
|
||
f"MCP {base_name} 的 URL 无效"
|
||
) from exc
|
||
if (
|
||
parsed_url.scheme not in {"http", "https"}
|
||
or not parsed_url.netloc
|
||
):
|
||
raise GrokBuildError(f"MCP {base_name} 的 URL 无效")
|
||
if parsed_url.fragment:
|
||
raise GrokBuildError(f"MCP {base_name} 的 URL 不能包含 fragment")
|
||
if parsed_url.username or parsed_url.password:
|
||
raise GrokBuildError(
|
||
f"MCP {base_name} 的 URL 不能内嵌用户名或密码"
|
||
)
|
||
sensitive_query_names = {
|
||
"api_key",
|
||
"apikey",
|
||
"access_token",
|
||
"authorization",
|
||
"auth",
|
||
"bearer",
|
||
"key",
|
||
"token",
|
||
"secret",
|
||
"password",
|
||
"credential",
|
||
"signature",
|
||
"sig",
|
||
}
|
||
if any(
|
||
key.strip().lower().replace("-", "_")
|
||
in sensitive_query_names
|
||
for key, _value in parse_qsl(
|
||
parsed_url.query,
|
||
keep_blank_values=True,
|
||
)
|
||
):
|
||
raise GrokBuildError(
|
||
f"MCP {base_name} 的 URL query 含疑似密钥；请移到 headers"
|
||
)
|
||
output.extend(["", f"[mcp_servers.{name}]"])
|
||
if transport == "stdio":
|
||
output.append(f"command = {_toml_string(command)}")
|
||
args = raw.get("args")
|
||
if isinstance(args, list):
|
||
if any(secret_option.search(str(item)) for item in args):
|
||
raise GrokBuildError(
|
||
f"MCP {base_name} 的 args 含疑似密钥参数；请移到 env"
|
||
)
|
||
encoded = ", ".join(_toml_string(item) for item in args)
|
||
output.append(f"args = [{encoded}]")
|
||
env = raw.get("env")
|
||
if isinstance(env, dict) and env:
|
||
entries = ", ".join(
|
||
f"{_toml_string(key)} = "
|
||
f"{_toml_string(self._mcp_config_value(name, 'env', key, value))}"
|
||
for key, value in env.items()
|
||
)
|
||
output.append(f"env = {{ {entries} }}")
|
||
cwd = str(raw.get("cwd") or "").strip()
|
||
if cwd:
|
||
output.append(f"cwd = {_toml_string(cwd)}")
|
||
else:
|
||
output.append(f"url = {_toml_string(url)}")
|
||
if transport == "sse":
|
||
output.append('type = "sse"')
|
||
headers = raw.get("headers")
|
||
if isinstance(headers, dict) and headers:
|
||
entries = ", ".join(
|
||
f"{_toml_string(key)} = "
|
||
f"{_toml_string(self._mcp_config_value(name, 'header', key, value))}"
|
||
for key, value in headers.items()
|
||
)
|
||
output.append(f"headers = {{ {entries} }}")
|
||
numeric_fields = (
|
||
"startup_timeout_sec",
|
||
"tool_timeout_sec",
|
||
)
|
||
for field in numeric_fields:
|
||
try:
|
||
number = int(raw.get(field))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if number > 0:
|
||
output.append(f"{field} = {number}")
|
||
tool_timeouts = raw.get("tool_timeouts")
|
||
if isinstance(tool_timeouts, dict) and tool_timeouts:
|
||
entries: list[str] = []
|
||
for key, value in tool_timeouts.items():
|
||
try:
|
||
timeout = int(value)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if timeout > 0:
|
||
entries.append(f"{_toml_string(key)} = {timeout}")
|
||
if entries:
|
||
output.append(f"tool_timeouts = {{ {', '.join(entries)} }}")
|
||
return output
|
||
|
||
@staticmethod
|
||
def _parse_toml_key_path(raw: str) -> tuple[str, ...] | None:
|
||
"""Parse bare/basic/literal TOML dotted keys used by table headers."""
|
||
parts: list[str] = []
|
||
cursor = 0
|
||
length = len(raw)
|
||
escapes = {
|
||
"b": "\b",
|
||
"t": "\t",
|
||
"n": "\n",
|
||
"f": "\f",
|
||
"r": "\r",
|
||
'"': '"',
|
||
"\\": "\\",
|
||
}
|
||
while True:
|
||
while cursor < length and raw[cursor].isspace():
|
||
cursor += 1
|
||
if cursor >= length:
|
||
return tuple(parts) if parts else None
|
||
if raw[cursor] == '"':
|
||
cursor += 1
|
||
value: list[str] = []
|
||
closed = False
|
||
while cursor < length:
|
||
character = raw[cursor]
|
||
if character == '"':
|
||
cursor += 1
|
||
closed = True
|
||
break
|
||
if character != "\\":
|
||
value.append(character)
|
||
cursor += 1
|
||
continue
|
||
cursor += 1
|
||
if cursor >= length:
|
||
return None
|
||
escape = raw[cursor]
|
||
if escape in escapes:
|
||
value.append(escapes[escape])
|
||
cursor += 1
|
||
continue
|
||
if escape in {"u", "U"}:
|
||
digits = 4 if escape == "u" else 8
|
||
encoded = raw[cursor + 1 : cursor + 1 + digits]
|
||
if (
|
||
len(encoded) != digits
|
||
or not re.fullmatch(r"[0-9A-Fa-f]+", encoded)
|
||
):
|
||
return None
|
||
value.append(chr(int(encoded, 16)))
|
||
cursor += 1 + digits
|
||
continue
|
||
return None
|
||
if not closed:
|
||
return None
|
||
part = "".join(value)
|
||
elif raw[cursor] == "'":
|
||
end = raw.find("'", cursor + 1)
|
||
if end < 0:
|
||
return None
|
||
part = raw[cursor + 1 : end]
|
||
cursor = end + 1
|
||
else:
|
||
match = re.match(r"[A-Za-z0-9_-]+", raw[cursor:])
|
||
if not match:
|
||
return None
|
||
part = match.group(0)
|
||
cursor += len(part)
|
||
parts.append(part)
|
||
while cursor < length and raw[cursor].isspace():
|
||
cursor += 1
|
||
if cursor >= length:
|
||
return tuple(parts)
|
||
if raw[cursor] != ".":
|
||
return None
|
||
cursor += 1
|
||
|
||
@staticmethod
|
||
def _toml_key(value: object) -> str:
|
||
rendered = str(value)
|
||
if re.fullmatch(r"[A-Za-z0-9_-]+", rendered):
|
||
return rendered
|
||
return _toml_string(rendered)
|
||
|
||
@classmethod
|
||
def _toml_value(cls, value: object) -> str:
|
||
"""Serialize values returned by tomllib without changing semantics."""
|
||
if isinstance(value, str):
|
||
return _toml_string(value)
|
||
if isinstance(value, bool):
|
||
return "true" if value else "false"
|
||
if isinstance(value, int):
|
||
return str(value)
|
||
if isinstance(value, float):
|
||
return repr(value)
|
||
if isinstance(value, (datetime, date, datetime_time)):
|
||
return value.isoformat()
|
||
if isinstance(value, list):
|
||
return "[" + ", ".join(cls._toml_value(item) for item in value) + "]"
|
||
if isinstance(value, dict):
|
||
entries = ", ".join(
|
||
f"{cls._toml_key(key)} = {cls._toml_value(item)}"
|
||
for key, item in value.items()
|
||
)
|
||
return f"{{ {entries} }}"
|
||
raise GrokBuildError(
|
||
f"Grok config.toml 含无法安全保留的 TOML 值类型：{type(value).__name__}"
|
||
)
|
||
|
||
@staticmethod
|
||
def _parse_toml_document(content: str, source: str = "Grok config.toml") -> dict:
|
||
if not content.strip():
|
||
return {}
|
||
try:
|
||
parsed = tomllib.loads(content)
|
||
except tomllib.TOMLDecodeError as exc:
|
||
raise GrokBuildError(f"{source} 不是有效 TOML：{exc}") from exc
|
||
if not isinstance(parsed, dict):
|
||
raise GrokBuildError(f"{source} 的根节点必须是 TOML 表")
|
||
return parsed
|
||
|
||
@classmethod
|
||
def _split_toml_assignment(
|
||
cls,
|
||
line: str,
|
||
) -> tuple[tuple[str, ...], int] | None:
|
||
"""Return a real assignment key and the value's starting column."""
|
||
cursor = 0
|
||
quote: str | None = None
|
||
while cursor < len(line):
|
||
character = line[cursor]
|
||
if quote == '"':
|
||
if character == "\\":
|
||
cursor += 2
|
||
continue
|
||
if character == '"':
|
||
quote = None
|
||
cursor += 1
|
||
continue
|
||
if quote == "'":
|
||
if character == "'":
|
||
quote = None
|
||
cursor += 1
|
||
continue
|
||
if character == "#":
|
||
return None
|
||
if character in {'"', "'"}:
|
||
quote = character
|
||
cursor += 1
|
||
continue
|
||
if character == "=":
|
||
path = cls._parse_toml_key_path(line[:cursor].strip())
|
||
return (path, cursor + 1) if path else None
|
||
cursor += 1
|
||
return None
|
||
|
||
@staticmethod
|
||
def _toml_value_end(
|
||
lines: Sequence[str],
|
||
start: int,
|
||
value_column: int,
|
||
) -> int:
|
||
"""Find the final line of one syntactically valid TOML value."""
|
||
multiline: str | None = None
|
||
quote: str | None = None
|
||
depth: list[str] = []
|
||
matching = {"]": "[", "}": "{"}
|
||
for line_index in range(start, len(lines)):
|
||
line = lines[line_index]
|
||
cursor = value_column if line_index == start else 0
|
||
while cursor < len(line):
|
||
if multiline is not None:
|
||
end = line.find(multiline, cursor)
|
||
if end < 0:
|
||
cursor = len(line)
|
||
continue
|
||
if multiline == '"""':
|
||
backslashes = 0
|
||
check = end - 1
|
||
while check >= 0 and line[check] == "\\":
|
||
backslashes += 1
|
||
check -= 1
|
||
if backslashes % 2:
|
||
cursor = end + 3
|
||
continue
|
||
multiline = None
|
||
cursor = end + 3
|
||
continue
|
||
if quote == '"':
|
||
if line[cursor] == "\\":
|
||
cursor += 2
|
||
elif line[cursor] == '"':
|
||
quote = None
|
||
cursor += 1
|
||
else:
|
||
cursor += 1
|
||
continue
|
||
if quote == "'":
|
||
if line[cursor] == "'":
|
||
quote = None
|
||
cursor += 1
|
||
continue
|
||
if line.startswith('"""', cursor):
|
||
multiline = '"""'
|
||
cursor += 3
|
||
continue
|
||
if line.startswith("'''", cursor):
|
||
multiline = "'''"
|
||
cursor += 3
|
||
continue
|
||
character = line[cursor]
|
||
if character == "#":
|
||
break
|
||
if character in {'"', "'"}:
|
||
quote = character
|
||
elif character in "[{":
|
||
depth.append(character)
|
||
elif character in "]}":
|
||
if depth and depth[-1] == matching[character]:
|
||
depth.pop()
|
||
cursor += 1
|
||
if multiline is None and quote is None and not depth:
|
||
return line_index
|
||
return len(lines) - 1
|
||
|
||
@classmethod
|
||
def _toml_assignments(
|
||
cls,
|
||
lines: Sequence[str],
|
||
outside: Sequence[bool],
|
||
headers: Mapping[int, tuple[str, tuple[str, ...] | None]],
|
||
) -> dict[int, tuple[int, tuple[str, ...]]]:
|
||
"""Map assignment start lines to their end line and semantic key path."""
|
||
assignments: dict[int, tuple[int, tuple[str, ...]]] = {}
|
||
current_table: tuple[str, ...] = ()
|
||
skip_until = -1
|
||
for index, line in enumerate(lines):
|
||
header = headers.get(index)
|
||
if header is not None:
|
||
_kind, section = header
|
||
current_table = section or ()
|
||
continue
|
||
if index <= skip_until or not outside[index]:
|
||
continue
|
||
assignment = cls._split_toml_assignment(line)
|
||
if assignment is None:
|
||
continue
|
||
key_path, value_column = assignment
|
||
end = cls._toml_value_end(lines, index, value_column)
|
||
assignments[index] = (end, current_table + key_path)
|
||
skip_until = end
|
||
return assignments
|
||
|
||
@classmethod
|
||
def _parse_toml_header(
|
||
cls,
|
||
line: str,
|
||
) -> tuple[str, tuple[str, ...] | None] | None:
|
||
"""Parse a table header while allowing brackets inside quoted keys."""
|
||
stripped = line.lstrip()
|
||
if not stripped.startswith("["):
|
||
return None
|
||
array = stripped.startswith("[[")
|
||
opening = 2 if array else 1
|
||
cursor = opening
|
||
quote: str | None = None
|
||
closing_start = -1
|
||
while cursor < len(stripped):
|
||
character = stripped[cursor]
|
||
if quote == '"':
|
||
if character == "\\":
|
||
cursor += 2
|
||
continue
|
||
if character == '"':
|
||
quote = None
|
||
cursor += 1
|
||
continue
|
||
if quote == "'":
|
||
if character == "'":
|
||
quote = None
|
||
cursor += 1
|
||
continue
|
||
if character in {'"', "'"}:
|
||
quote = character
|
||
cursor += 1
|
||
continue
|
||
if array and stripped.startswith("]]", cursor):
|
||
closing_start = cursor
|
||
cursor += 2
|
||
break
|
||
if not array and character == "]":
|
||
closing_start = cursor
|
||
cursor += 1
|
||
break
|
||
cursor += 1
|
||
if closing_start < 0 or quote is not None:
|
||
return None
|
||
remainder = stripped[cursor:].strip()
|
||
if remainder and not remainder.startswith("#"):
|
||
return None
|
||
raw_path = stripped[opening:closing_start]
|
||
return (
|
||
"array" if array else "table",
|
||
cls._parse_toml_key_path(raw_path),
|
||
)
|
||
|
||
@classmethod
|
||
def _toml_structure(
|
||
cls,
|
||
lines: Sequence[str],
|
||
) -> tuple[
|
||
list[bool],
|
||
dict[int, tuple[str, tuple[str, ...] | None]],
|
||
]:
|
||
"""Locate real TOML table headers without matching multiline strings."""
|
||
outside_at_start: list[bool] = []
|
||
headers: dict[int, tuple[str, tuple[str, ...] | None]] = {}
|
||
multiline: str | None = None
|
||
containers: list[str] = []
|
||
matching = {"]": "[", "}": "{"}
|
||
|
||
def escaped(text: str, position: int) -> bool:
|
||
backslashes = 0
|
||
cursor = position - 1
|
||
while cursor >= 0 and text[cursor] == "\\":
|
||
backslashes += 1
|
||
cursor -= 1
|
||
return backslashes % 2 == 1
|
||
|
||
for index, line in enumerate(lines):
|
||
starts_outside = multiline is None and not containers
|
||
outside_at_start.append(starts_outside)
|
||
if starts_outside:
|
||
header = cls._parse_toml_header(line)
|
||
if header is not None:
|
||
headers[index] = header
|
||
|
||
cursor = 0
|
||
length = len(line)
|
||
while cursor < length:
|
||
if multiline is not None:
|
||
delimiter = multiline
|
||
end = line.find(delimiter, cursor)
|
||
while (
|
||
end >= 0
|
||
and delimiter == '"""'
|
||
and escaped(line, end)
|
||
):
|
||
end = line.find(delimiter, end + 3)
|
||
if end < 0:
|
||
break
|
||
multiline = None
|
||
cursor = end + 3
|
||
continue
|
||
|
||
if line.startswith('"""', cursor):
|
||
multiline = '"""'
|
||
cursor += 3
|
||
continue
|
||
if line.startswith("'''", cursor):
|
||
multiline = "'''"
|
||
cursor += 3
|
||
continue
|
||
character = line[cursor]
|
||
if character == "#":
|
||
break
|
||
if character == '"':
|
||
cursor += 1
|
||
while cursor < length:
|
||
if line[cursor] == "\\":
|
||
cursor += 2
|
||
elif line[cursor] == '"':
|
||
cursor += 1
|
||
break
|
||
else:
|
||
cursor += 1
|
||
continue
|
||
if character == "'":
|
||
end = line.find("'", cursor + 1)
|
||
cursor = length if end < 0 else end + 1
|
||
continue
|
||
if character in "[{":
|
||
containers.append(character)
|
||
elif (
|
||
character in "]}"
|
||
and containers
|
||
and containers[-1] == matching[character]
|
||
):
|
||
containers.pop()
|
||
cursor += 1
|
||
return outside_at_start, headers
|
||
|
||
@classmethod
|
||
def _without_managed_runtime_sections(
|
||
cls,
|
||
content: str,
|
||
*,
|
||
remove_owned_sections: bool = True,
|
||
remove_permission: bool = False,
|
||
) -> str:
|
||
"""Remove the managed block and optionally other bridge-owned tables."""
|
||
cls._parse_toml_document(content)
|
||
lines = content.splitlines()
|
||
outside, headers = cls._toml_structure(lines)
|
||
assignments = cls._toml_assignments(lines, outside, headers)
|
||
output: list[str] = []
|
||
skip_section = False
|
||
skip_assignment_until = -1
|
||
in_managed_block = False
|
||
begin_count = sum(
|
||
outside[index] and line.strip() == MANAGED_CONFIG_BEGIN
|
||
for index, line in enumerate(lines)
|
||
)
|
||
end_count = sum(
|
||
outside[index] and line.strip() == MANAGED_CONFIG_END
|
||
for index, line in enumerate(lines)
|
||
)
|
||
if begin_count != end_count or begin_count > 1:
|
||
raise GrokBuildError(
|
||
"Grok config.toml 中的企业微信 RPA 自动配置区块标记不唯一或不完整"
|
||
)
|
||
for index, line in enumerate(lines):
|
||
if index <= skip_assignment_until:
|
||
continue
|
||
stripped = line.strip()
|
||
if outside[index] and stripped == MANAGED_CONFIG_BEGIN:
|
||
if in_managed_block:
|
||
raise GrokBuildError("Grok 自动配置区块出现嵌套起始标记")
|
||
in_managed_block = True
|
||
continue
|
||
if outside[index] and stripped == MANAGED_CONFIG_END:
|
||
if not in_managed_block:
|
||
raise GrokBuildError("Grok 自动配置区块缺少起始标记")
|
||
in_managed_block = False
|
||
continue
|
||
if in_managed_block:
|
||
continue
|
||
|
||
header = headers.get(index)
|
||
if header is not None:
|
||
kind, section = header
|
||
skip_section = (
|
||
remove_owned_sections
|
||
and section is not None
|
||
and (
|
||
section[:1] == ("models",)
|
||
or section[:1] == ("model",)
|
||
or section[:1] == ("ui",)
|
||
or section[:1] == ("suggestions",)
|
||
or section[:1] == ("subagents",)
|
||
or section[:1] == ("goal",)
|
||
or section[:1] == ("auto_mode",)
|
||
or section[:2] == ("compaction", "memory_flush")
|
||
or section[:1] == ("compat",)
|
||
or section[:1] == ("claude_compat",)
|
||
or section[:1] == ("plugins",)
|
||
or (
|
||
remove_permission
|
||
and section[:1] == ("permission",)
|
||
)
|
||
)
|
||
)
|
||
assignment = assignments.get(index)
|
||
if assignment is not None:
|
||
end, path = assignment
|
||
if (
|
||
remove_owned_sections
|
||
and (
|
||
path[:1] == ("models",)
|
||
or path[:1] == ("model",)
|
||
or path[:1] == ("ui",)
|
||
or path[:1] == ("suggestions",)
|
||
or path[:1] == ("subagents",)
|
||
or path[:1] == ("goal",)
|
||
or path[:1] == ("auto_mode",)
|
||
or path[:2] == ("compaction", "memory_flush")
|
||
or path[:1] == ("compat",)
|
||
or path[:1] == ("claude_compat",)
|
||
or path[:1] == ("plugins",)
|
||
or (
|
||
remove_permission
|
||
and path[:1] == ("permission",)
|
||
)
|
||
)
|
||
):
|
||
skip_assignment_until = end
|
||
continue
|
||
if not skip_section:
|
||
output.append(line)
|
||
if in_managed_block:
|
||
raise GrokBuildError(
|
||
"Grok config.toml 中的企业微信 RPA 自动配置区块不完整，请修复区块标记"
|
||
)
|
||
return "\n".join(output).strip()
|
||
|
||
@classmethod
|
||
def _models_table_extras(cls, content: str) -> list[str]:
|
||
"""Keep non-routing ``models`` settings while Agent model keys are owned.
|
||
|
||
Canonicalizing this one small namespace avoids corrupting legal TOML
|
||
that uses dotted keys, escaped quoted keys, or multiline values.
|
||
"""
|
||
parsed = cls._parse_toml_document(content)
|
||
models = parsed.get("models")
|
||
if not isinstance(models, dict):
|
||
return []
|
||
return [
|
||
f"{cls._toml_key(key)} = {cls._toml_value(value)}"
|
||
for key, value in models.items()
|
||
if str(key) not in MANAGED_MODELS_KEYS
|
||
]
|
||
|
||
@classmethod
|
||
def _ui_table_extras(cls, content: str) -> list[str]:
|
||
"""Preserve visual UI preferences but never model-routing UI keys."""
|
||
parsed = cls._parse_toml_document(content)
|
||
ui = parsed.get("ui")
|
||
if not isinstance(ui, dict):
|
||
return []
|
||
return [
|
||
f"{cls._toml_key(key)} = {cls._toml_value(value)}"
|
||
for key, value in ui.items()
|
||
if str(key) not in MANAGED_UI_KEYS
|
||
]
|
||
|
||
@classmethod
|
||
def _nested_table_extras(
|
||
cls,
|
||
content: str,
|
||
path: Sequence[str],
|
||
excluded: set[str] | frozenset[str],
|
||
) -> list[str]:
|
||
parsed: object = cls._parse_toml_document(content)
|
||
for component in path:
|
||
if not isinstance(parsed, dict):
|
||
return []
|
||
parsed = parsed.get(component)
|
||
if not isinstance(parsed, dict):
|
||
return []
|
||
return [
|
||
f"{cls._toml_key(key)} = {cls._toml_value(value)}"
|
||
for key, value in parsed.items()
|
||
if str(key) not in excluded
|
||
]
|
||
|
||
@classmethod
|
||
def _plugin_disabled_names(cls, content: str) -> set[str]:
|
||
parsed = cls._parse_toml_document(content)
|
||
plugins = parsed.get("plugins")
|
||
if not isinstance(plugins, dict):
|
||
return set()
|
||
disabled = plugins.get("disabled")
|
||
if not isinstance(disabled, list):
|
||
return set()
|
||
return {
|
||
str(value).strip()
|
||
for value in disabled
|
||
if isinstance(value, str) and str(value).strip()
|
||
}
|
||
|
||
def _claude_compat_plugin_names(self) -> set[str]:
|
||
"""Return Claude-installed plugin names that Grok would auto-discover."""
|
||
manifest = (
|
||
self.user_home
|
||
/ ".claude"
|
||
/ "plugins"
|
||
/ "installed_plugins.json"
|
||
)
|
||
try:
|
||
payload = json.loads(manifest.read_text(encoding="utf-8"))
|
||
except (FileNotFoundError, OSError, ValueError, TypeError):
|
||
return set()
|
||
plugins = payload.get("plugins") if isinstance(payload, dict) else None
|
||
if not isinstance(plugins, dict):
|
||
return set()
|
||
names: set[str] = set()
|
||
for identifier, installations in plugins.items():
|
||
plugin_id = str(identifier).strip()
|
||
if plugin_id:
|
||
names.add(plugin_id)
|
||
continue
|
||
if not isinstance(installations, list):
|
||
continue
|
||
for installation in installations:
|
||
if not isinstance(installation, dict):
|
||
continue
|
||
install_path = str(
|
||
installation.get("installPath") or ""
|
||
).strip()
|
||
if not install_path:
|
||
continue
|
||
candidate = Path(install_path).parent.name.strip()
|
||
if candidate:
|
||
names.add(candidate)
|
||
return names
|
||
|
||
def _external_compat_mcp_names(self) -> set[str]:
|
||
"""Return MCP names discovered from disabled Cursor/Claude sources."""
|
||
candidates = (
|
||
self.user_home / ".cursor" / "mcp.json",
|
||
self.user_home / ".claude.json",
|
||
self.project_dir / ".cursor" / "mcp.json",
|
||
self.project_dir / ".mcp.json",
|
||
)
|
||
names: set[str] = set()
|
||
|
||
def collect(value: object) -> None:
|
||
if isinstance(value, dict):
|
||
servers = value.get("mcpServers")
|
||
if isinstance(servers, dict):
|
||
names.update(
|
||
str(name).strip()
|
||
for name in servers
|
||
if str(name).strip()
|
||
)
|
||
for child in value.values():
|
||
collect(child)
|
||
elif isinstance(value, list):
|
||
for child in value:
|
||
collect(child)
|
||
|
||
for path in candidates:
|
||
try:
|
||
collect(json.loads(path.read_text(encoding="utf-8")))
|
||
except (FileNotFoundError, OSError, ValueError, TypeError):
|
||
continue
|
||
return names
|
||
|
||
@classmethod
|
||
def _configured_mcp_names(cls, content: str) -> set[str]:
|
||
parsed = cls._parse_toml_document(content)
|
||
servers = parsed.get("mcp_servers")
|
||
if not isinstance(servers, dict):
|
||
return set()
|
||
return {str(name).strip() for name in servers if str(name).strip()}
|
||
|
||
@classmethod
|
||
def _subagents_table_extras(cls, content: str) -> list[str]:
|
||
"""Keep subagent behavior while rewriting role/persona model pins."""
|
||
parsed = cls._parse_toml_document(content)
|
||
subagents = parsed.get("subagents")
|
||
if not isinstance(subagents, dict):
|
||
return []
|
||
preserved = copy.deepcopy(subagents)
|
||
preserved.pop("enabled", None)
|
||
preserved.pop("models", None)
|
||
for collection_name in ("roles", "personas"):
|
||
definitions = preserved.get(collection_name)
|
||
if not isinstance(definitions, dict):
|
||
continue
|
||
for definition in definitions.values():
|
||
if isinstance(definition, dict) and "model" in definition:
|
||
definition["model"] = MODEL_PROFILE
|
||
return [
|
||
f"{cls._toml_key(key)} = {cls._toml_value(value)}"
|
||
for key, value in preserved.items()
|
||
]
|
||
|
||
def _subagent_pin_names(self, content: str) -> set[str]:
|
||
names = set(PINNED_SUBAGENT_NAMES)
|
||
parsed = self._parse_toml_document(content)
|
||
subagents = parsed.get("subagents")
|
||
if isinstance(subagents, dict):
|
||
for field in ("models", "toggle", "roles", "personas"):
|
||
values = subagents.get(field)
|
||
if isinstance(values, dict):
|
||
names.update(str(key) for key in values)
|
||
discovery_roots = (
|
||
self.runtime_home,
|
||
self.project_dir / ".grok",
|
||
)
|
||
for root in discovery_roots:
|
||
for folder_name in ("roles", "personas", "agents"):
|
||
folder = root / folder_name
|
||
try:
|
||
entries = tuple(folder.iterdir())
|
||
except OSError:
|
||
continue
|
||
for entry in entries:
|
||
if not entry.is_file() or entry.suffix.lower() not in {
|
||
".toml",
|
||
".md",
|
||
}:
|
||
continue
|
||
names.add(entry.stem)
|
||
if entry.suffix.lower() != ".md":
|
||
continue
|
||
try:
|
||
header = entry.read_text(
|
||
encoding="utf-8", errors="replace"
|
||
)[:16384]
|
||
except OSError:
|
||
continue
|
||
match = re.search(
|
||
r"(?mi)^\s*name\s*:\s*['\"]?([^'\"\r\n#]+)",
|
||
header,
|
||
)
|
||
if match:
|
||
names.add(match.group(1).strip())
|
||
return {name for name in names if name.strip()}
|
||
|
||
def _write_runtime_config(self, managed_content: str) -> bool:
|
||
try:
|
||
existing = self.user_config_file.read_text(encoding="utf-8")
|
||
except FileNotFoundError:
|
||
existing = ""
|
||
except OSError as exc:
|
||
raise GrokBuildError(f"无法读取 Grok Build 用户配置：{exc}") from exc
|
||
unmanaged_existing = self._without_managed_runtime_sections(
|
||
existing,
|
||
remove_owned_sections=False,
|
||
)
|
||
table_extras = {
|
||
"models": self._models_table_extras(existing),
|
||
"ui": self._ui_table_extras(existing),
|
||
"suggestions": self._nested_table_extras(
|
||
existing,
|
||
("suggestions",),
|
||
{"enabled", "ai_enabled", "ai_model"},
|
||
),
|
||
"subagents": self._subagents_table_extras(existing),
|
||
"goal": self._nested_table_extras(
|
||
existing,
|
||
("goal",),
|
||
{
|
||
"use_current_model_only",
|
||
"planner_model",
|
||
"strategist_model",
|
||
"skeptic_models",
|
||
},
|
||
),
|
||
"auto_mode": self._nested_table_extras(
|
||
existing,
|
||
("auto_mode",),
|
||
{"classifier_model"},
|
||
),
|
||
"compaction.memory_flush": self._nested_table_extras(
|
||
existing,
|
||
("compaction", "memory_flush"),
|
||
{"flush_model"},
|
||
),
|
||
"plugins": self._nested_table_extras(
|
||
unmanaged_existing,
|
||
("plugins",),
|
||
{"disabled"},
|
||
),
|
||
}
|
||
preserved = self._without_managed_runtime_sections(
|
||
existing,
|
||
remove_permission="[permission]" in managed_content,
|
||
)
|
||
additional_subagents = sorted(
|
||
self._subagent_pin_names(existing) - PINNED_SUBAGENT_NAMES
|
||
)
|
||
if additional_subagents and "[subagents.models]" in managed_content:
|
||
pins = "\n".join(
|
||
f"{self._toml_key(name)} = {_toml_string(MODEL_PROFILE)}"
|
||
for name in additional_subagents
|
||
)
|
||
managed_content = managed_content.replace(
|
||
"[subagents.models]",
|
||
f"[subagents.models]\n{pins}",
|
||
1,
|
||
)
|
||
for table_name, extras in table_extras.items():
|
||
if not extras:
|
||
continue
|
||
extra_text = "\n".join(extras).strip()
|
||
table_header = f"[{table_name}]"
|
||
if table_header in managed_content:
|
||
managed_content = managed_content.replace(
|
||
table_header,
|
||
f"{table_header}\n{extra_text}",
|
||
1,
|
||
)
|
||
else:
|
||
managed_content = managed_content.replace(
|
||
MANAGED_CONFIG_END,
|
||
f"{table_header}\n{extra_text}\n\n{MANAGED_CONFIG_END}",
|
||
1,
|
||
)
|
||
pieces = [value for value in (preserved, managed_content.strip()) if value]
|
||
rendered = "\n\n".join(pieces) + "\n"
|
||
self._parse_toml_document(rendered, "生成后的 Grok config.toml")
|
||
changed = not hmac.compare_digest(
|
||
existing.encode("utf-8"),
|
||
rendered.encode("utf-8"),
|
||
)
|
||
if changed:
|
||
self._atomic_write(self.user_config_file, rendered)
|
||
|
||
# Migrate only the legacy file written by older bridge revisions. The
|
||
# filename is reserved by Grok for remotely served enterprise policy.
|
||
try:
|
||
legacy = self.legacy_managed_config_file.read_text(encoding="utf-8")
|
||
except (FileNotFoundError, OSError):
|
||
legacy = ""
|
||
if legacy.startswith("# 由企业微信 RPA"):
|
||
self.legacy_managed_config_file.unlink(missing_ok=True)
|
||
return changed
|
||
|
||
def sync_model_configuration(
|
||
self,
|
||
include_mcp: bool | None = None,
|
||
) -> ModelSyncResult:
|
||
with self._runtime_sync_lock:
|
||
return self._sync_model_configuration_locked(include_mcp)
|
||
|
||
def _sync_model_configuration_locked(
|
||
self,
|
||
include_mcp: bool | None = None,
|
||
) -> ModelSyncResult:
|
||
ai_settings = self.load_ai_settings()
|
||
integration = self.load_integration_settings()
|
||
if include_mcp is None:
|
||
include_mcp = bool(integration.get("sync_mcp_servers", False))
|
||
external_compatibility = bool(
|
||
integration.get("external_compatibility", False)
|
||
)
|
||
try:
|
||
existing_config = self.user_config_file.read_text(encoding="utf-8")
|
||
except (FileNotFoundError, OSError):
|
||
existing_config = ""
|
||
unmanaged_config = self._without_managed_runtime_sections(
|
||
existing_config,
|
||
remove_owned_sections=False,
|
||
)
|
||
disabled_plugins = self._plugin_disabled_names(unmanaged_config)
|
||
disabled_external_mcp_names: set[str] = set()
|
||
if not external_compatibility:
|
||
disabled_plugins.update(self._claude_compat_plugin_names())
|
||
disabled_external_mcp_names = (
|
||
self._external_compat_mcp_names()
|
||
- self._configured_mcp_names(unmanaged_config)
|
||
)
|
||
profile = self.agent_model_profile(ai_settings)
|
||
self.runtime_home.mkdir(parents=True, exist_ok=True)
|
||
include_customer_service_tools = bool(
|
||
integration.get("customer_service_tools", True)
|
||
)
|
||
content = self._render_managed_config(
|
||
profile,
|
||
ai_settings,
|
||
bool(include_mcp),
|
||
include_customer_service_tools,
|
||
external_compatibility=external_compatibility,
|
||
disabled_plugins=tuple(sorted(disabled_plugins)),
|
||
disabled_external_mcp_names=tuple(
|
||
sorted(disabled_external_mcp_names)
|
||
),
|
||
)
|
||
config_changed = self._write_runtime_config(content)
|
||
display_backend = profile.source_backend or profile.api_backend
|
||
display_base_url = profile.source_base_url or profile.base_url
|
||
if profile.compatible:
|
||
if profile.source_backend == "dify":
|
||
message = (
|
||
f"已通过本地工具调用适配器配置 {profile.model}"
|
||
"（Dify Chat Messages）"
|
||
)
|
||
else:
|
||
message = f"已自动配置 {profile.model}（{profile.api_backend}）"
|
||
else:
|
||
message = f"未启用后台模型：{profile.reason}"
|
||
result = ModelSyncResult(
|
||
compatible=profile.compatible,
|
||
configured=profile.compatible,
|
||
profile=profile.profile,
|
||
model=profile.model,
|
||
base_url=display_base_url,
|
||
api_backend=display_backend,
|
||
config_path=str(self.managed_config_file),
|
||
synced_at=_utc_timestamp(),
|
||
message=message,
|
||
source_base_url=display_base_url,
|
||
source_api_backend=display_backend,
|
||
effective_base_url=profile.base_url,
|
||
effective_api_backend=profile.api_backend,
|
||
adapter_instance_id=profile.adapter_instance_id,
|
||
)
|
||
previous = self.read_sync_result()
|
||
previous_values = asdict(previous)
|
||
current_values = asdict(result)
|
||
previous_values.pop("synced_at", None)
|
||
current_values.pop("synced_at", None)
|
||
if (
|
||
not config_changed
|
||
and bool(previous.synced_at)
|
||
and previous_values == current_values
|
||
):
|
||
return previous
|
||
self._atomic_write(
|
||
self.sync_state_file,
|
||
json.dumps(asdict(result), ensure_ascii=False, indent=2) + "\n",
|
||
)
|
||
return result
|
||
|
||
def prepare_agent_configuration(
|
||
self,
|
||
include_mcp: bool | None = None,
|
||
*,
|
||
verify_endpoint: bool = True,
|
||
probe_timeout: float = 12.0,
|
||
) -> ModelSyncResult:
|
||
"""Refresh process-bound endpoints before launching an Agent process."""
|
||
result = self.sync_model_configuration(include_mcp=include_mcp)
|
||
if not result.compatible:
|
||
raise GrokBuildError(
|
||
result.message
|
||
or "后台 Agent 自有模型配置不兼容，已阻止启动"
|
||
)
|
||
if verify_endpoint:
|
||
probe = self.probe_agent_model(
|
||
force=False,
|
||
timeout=probe_timeout,
|
||
cache_ttl=30.0,
|
||
)
|
||
if not probe.ok:
|
||
raise GrokBuildError(
|
||
probe.message
|
||
or "后台 Agent 自有模型端点或工具调用协议预检失败"
|
||
)
|
||
return result
|
||
|
||
def read_sync_result(self) -> ModelSyncResult:
|
||
try:
|
||
raw = json.loads(self.sync_state_file.read_text(encoding="utf-8"))
|
||
return ModelSyncResult(
|
||
compatible=bool(raw.get("compatible")),
|
||
configured=bool(raw.get("configured")),
|
||
profile=str(raw.get("profile") or MODEL_PROFILE),
|
||
model=str(raw.get("model") or ""),
|
||
base_url=str(raw.get("base_url") or ""),
|
||
api_backend=str(raw.get("api_backend") or ""),
|
||
config_path=str(raw.get("config_path") or self.managed_config_file),
|
||
synced_at=str(raw.get("synced_at") or ""),
|
||
message=str(raw.get("message") or "尚未同步模型配置"),
|
||
source_base_url=str(
|
||
raw.get("source_base_url") or raw.get("base_url") or ""
|
||
),
|
||
source_api_backend=str(
|
||
raw.get("source_api_backend")
|
||
or raw.get("api_backend")
|
||
or ""
|
||
),
|
||
effective_base_url=str(raw.get("effective_base_url") or ""),
|
||
effective_api_backend=str(
|
||
raw.get("effective_api_backend") or ""
|
||
),
|
||
adapter_instance_id=str(
|
||
raw.get("adapter_instance_id") or ""
|
||
),
|
||
)
|
||
except (OSError, ValueError, TypeError):
|
||
return ModelSyncResult(
|
||
compatible=False,
|
||
configured=False,
|
||
profile=MODEL_PROFILE,
|
||
model="",
|
||
base_url="",
|
||
api_backend="",
|
||
config_path=str(self.managed_config_file),
|
||
synced_at="",
|
||
message="尚未同步模型配置",
|
||
)
|
||
|
||
def _managed_config_block(self) -> str:
|
||
try:
|
||
content = self.user_config_file.read_text(encoding="utf-8")
|
||
except (FileNotFoundError, OSError):
|
||
return ""
|
||
lines = content.splitlines()
|
||
outside, _headers = self._toml_structure(lines)
|
||
begins = [
|
||
index
|
||
for index, line in enumerate(lines)
|
||
if outside[index] and line.strip() == MANAGED_CONFIG_BEGIN
|
||
]
|
||
ends = [
|
||
index
|
||
for index, line in enumerate(lines)
|
||
if outside[index] and line.strip() == MANAGED_CONFIG_END
|
||
]
|
||
if len(begins) != 1 or len(ends) != 1 or begins[0] >= ends[0]:
|
||
return ""
|
||
return "\n".join(lines[begins[0] : ends[0] + 1])
|
||
|
||
@staticmethod
|
||
def _toml_references_environment(value: object, variable: str) -> bool:
|
||
if isinstance(value, str):
|
||
return value == variable or f"${{{variable}}}" in value
|
||
if isinstance(value, list):
|
||
return any(
|
||
GrokBuildManager._toml_references_environment(item, variable)
|
||
for item in value
|
||
)
|
||
if isinstance(value, dict):
|
||
return any(
|
||
GrokBuildManager._toml_references_environment(item, variable)
|
||
for item in value.values()
|
||
)
|
||
return False
|
||
|
||
@staticmethod
|
||
def _expected_model_table(profile: ModelProfile) -> dict[str, object]:
|
||
expected: dict[str, object] = {
|
||
"model": profile.model,
|
||
"base_url": profile.base_url,
|
||
"name": f"后台模型 · {profile.model}",
|
||
"env_key": MODEL_API_KEY_ENV,
|
||
"api_backend": profile.api_backend,
|
||
"temperature": profile.temperature,
|
||
"max_completion_tokens": profile.max_completion_tokens,
|
||
"context_window": profile.context_window,
|
||
}
|
||
if profile.source_backend != "dify":
|
||
expected["auth_scheme"] = profile.auth_scheme
|
||
if profile.api_backend == "messages" and profile.auth_scheme == "x_api_key":
|
||
expected["extra_headers"] = {"anthropic-version": "2023-06-01"}
|
||
return expected
|
||
|
||
def _custom_model_routes_are_locked(
|
||
self,
|
||
parsed_config: Mapping[str, object],
|
||
config_content: str,
|
||
) -> bool:
|
||
"""Confirm every built-in Agent model role resolves to our one profile."""
|
||
models = parsed_config.get("models")
|
||
ui = parsed_config.get("ui")
|
||
model_tables = parsed_config.get("model")
|
||
suggestions = parsed_config.get("suggestions")
|
||
subagents = parsed_config.get("subagents")
|
||
goal = parsed_config.get("goal")
|
||
auto_mode = parsed_config.get("auto_mode")
|
||
compaction = parsed_config.get("compaction")
|
||
memory_flush = (
|
||
compaction.get("memory_flush")
|
||
if isinstance(compaction, dict)
|
||
else None
|
||
)
|
||
subagent_models = (
|
||
subagents.get("models")
|
||
if isinstance(subagents, dict)
|
||
else None
|
||
)
|
||
required_subagent_names = self._subagent_pin_names(config_content)
|
||
definitions_safe = True
|
||
if isinstance(subagents, dict):
|
||
for collection_name in ("roles", "personas"):
|
||
definitions = subagents.get(collection_name)
|
||
if not isinstance(definitions, dict):
|
||
continue
|
||
for definition in definitions.values():
|
||
if not isinstance(definition, dict):
|
||
continue
|
||
model = definition.get("model")
|
||
if model is not None and not (
|
||
isinstance(model, str)
|
||
and model.strip() in {"", "inherit", MODEL_PROFILE}
|
||
):
|
||
definitions_safe = False
|
||
break
|
||
return bool(
|
||
isinstance(models, dict)
|
||
and models.get("default") == MODEL_PROFILE
|
||
and models.get("allowed_models") == [MODEL_PROFILE]
|
||
and models.get("web_search") == MODEL_PROFILE
|
||
and models.get("session_summary") == MODEL_PROFILE
|
||
and models.get("image_description") == MODEL_PROFILE
|
||
and models.get("prompt_suggestion") == MODEL_PROFILE
|
||
and isinstance(ui, dict)
|
||
and ui.get("prompt_suggestions") is False
|
||
and ui.get("fork_secondary_model") == MODEL_PROFILE
|
||
and isinstance(model_tables, dict)
|
||
and set(str(key) for key in model_tables) == {MODEL_PROFILE}
|
||
and isinstance(suggestions, dict)
|
||
and suggestions.get("enabled") is False
|
||
and suggestions.get("ai_enabled") is False
|
||
and suggestions.get("ai_model") == MODEL_PROFILE
|
||
and isinstance(subagents, dict)
|
||
and subagents.get("enabled") is True
|
||
and isinstance(subagent_models, dict)
|
||
and required_subagent_names.issubset(
|
||
{str(key) for key in subagent_models}
|
||
)
|
||
and all(
|
||
value == MODEL_PROFILE for value in subagent_models.values()
|
||
)
|
||
and definitions_safe
|
||
and isinstance(goal, dict)
|
||
and goal.get("use_current_model_only") is True
|
||
and not {
|
||
"planner_model",
|
||
"strategist_model",
|
||
"skeptic_models",
|
||
}.intersection(goal)
|
||
and isinstance(auto_mode, dict)
|
||
and auto_mode.get("classifier_model") == MODEL_PROFILE
|
||
and isinstance(memory_flush, dict)
|
||
and memory_flush.get("flush_model") == MODEL_PROFILE
|
||
)
|
||
|
||
def _agent_definition_model_violations(self, workspace: Path) -> list[str]:
|
||
"""Find file-based roles/personas/agents that pin another model."""
|
||
files: set[Path] = set()
|
||
grok_roots = {
|
||
self.runtime_home,
|
||
self.project_dir / ".grok",
|
||
workspace / ".grok",
|
||
}
|
||
for root in grok_roots:
|
||
for folder_name in ("roles", "personas", "agents"):
|
||
folder = root / folder_name
|
||
try:
|
||
files.update(
|
||
entry
|
||
for entry in folder.iterdir()
|
||
if entry.is_file()
|
||
and entry.suffix.lower() in {".toml", ".md"}
|
||
)
|
||
except OSError:
|
||
continue
|
||
plugins = root / "plugins"
|
||
if plugins.is_dir():
|
||
try:
|
||
files.update(
|
||
entry
|
||
for entry in plugins.rglob("*")
|
||
if entry.is_file()
|
||
and entry.parent.name.lower()
|
||
in {"roles", "personas", "agents"}
|
||
and entry.suffix.lower() in {".toml", ".md"}
|
||
)
|
||
except OSError:
|
||
pass
|
||
integration = self.load_integration_settings()
|
||
if bool(integration.get("external_compatibility", False)):
|
||
for folder in (
|
||
workspace / ".claude" / "agents",
|
||
self.user_home / ".claude" / "agents",
|
||
):
|
||
try:
|
||
files.update(
|
||
entry
|
||
for entry in folder.iterdir()
|
||
if entry.is_file() and entry.suffix.lower() == ".md"
|
||
)
|
||
except OSError:
|
||
continue
|
||
|
||
violations: list[str] = []
|
||
accepted = {"", "inherit", MODEL_PROFILE}
|
||
for path in sorted(files, key=lambda item: str(item).casefold()):
|
||
model: object = None
|
||
try:
|
||
content = path.read_text(
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
)
|
||
except OSError:
|
||
continue
|
||
if path.suffix.lower() == ".toml":
|
||
try:
|
||
parsed = tomllib.loads(content)
|
||
except (tomllib.TOMLDecodeError, ValueError, TypeError):
|
||
continue
|
||
model = parsed.get("model") if isinstance(parsed, dict) else None
|
||
else:
|
||
lines = content.splitlines()
|
||
if not lines or lines[0].strip() != "---":
|
||
continue
|
||
frontmatter: list[str] = []
|
||
for line in lines[1:]:
|
||
if line.strip() == "---":
|
||
break
|
||
frontmatter.append(line)
|
||
match = re.search(
|
||
r"(?mi)^\s*model\s*:\s*([^#\r\n]+)",
|
||
"\n".join(frontmatter),
|
||
)
|
||
if match:
|
||
model = match.group(1).strip().strip("'\"")
|
||
if model is None:
|
||
continue
|
||
normalized = model.strip() if isinstance(model, str) else ""
|
||
if not isinstance(model, str) or normalized not in accepted:
|
||
violations.append(f"{path} -> {model!r}")
|
||
return violations
|
||
|
||
def _assert_effective_config_isolated(
|
||
self,
|
||
environment: Mapping[str, str],
|
||
workspace: Path,
|
||
expected_user_config: str,
|
||
) -> None:
|
||
"""Verify that no higher-priority Grok layer can redirect credentials."""
|
||
requirements_file = self.runtime_home / "requirements.toml"
|
||
extra_known_layers = [
|
||
path
|
||
for path in (requirements_file, self.legacy_managed_config_file)
|
||
if path.exists()
|
||
]
|
||
if extra_known_layers:
|
||
raise GrokBuildError(
|
||
"检测到可能覆盖受管模型的 Grok 高优先级配置层："
|
||
+ "、".join(str(path) for path in extra_known_layers)
|
||
+ "；已阻止凭据注入"
|
||
)
|
||
|
||
binary = self.locate_binary()
|
||
if binary is None:
|
||
# The returned mapping cannot start Grok until a runtime exists.
|
||
# Every bridge launch path calls require_binary() before this point.
|
||
return
|
||
self.validate_binary(binary)
|
||
try:
|
||
completed = subprocess.run(
|
||
[str(binary), "--no-auto-update", "inspect", "--json"],
|
||
cwd=str(workspace),
|
||
env=dict(environment),
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.STDOUT,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
timeout=30,
|
||
check=False,
|
||
)
|
||
except (OSError, subprocess.SubprocessError) as exc:
|
||
raise GrokBuildError(
|
||
f"无法核验 Grok 实际配置层；已阻止凭据注入：{exc}"
|
||
) from exc
|
||
if completed.returncode != 0:
|
||
raise GrokBuildError(
|
||
"Grok inspect 无法核验实际配置层；已阻止凭据注入："
|
||
f"{(completed.stdout or '').strip()}"
|
||
)
|
||
try:
|
||
inspection = json.loads((completed.stdout or "").lstrip("\ufeff"))
|
||
except json.JSONDecodeError as exc:
|
||
raise GrokBuildError(
|
||
"Grok inspect 未返回有效配置层 JSON；已阻止凭据注入"
|
||
) from exc
|
||
config_sources = (
|
||
inspection.get("configSources") if isinstance(inspection, dict) else None
|
||
)
|
||
layers = (
|
||
config_sources.get("layers")
|
||
if isinstance(config_sources, dict)
|
||
else None
|
||
)
|
||
if not isinstance(layers, list):
|
||
raise GrokBuildError(
|
||
"Grok inspect 未提供可验证的 configSources.layers；"
|
||
"已阻止凭据注入"
|
||
)
|
||
expected_path = os.path.normcase(
|
||
os.path.abspath(str(self.user_config_file))
|
||
)
|
||
normalized_layers: list[tuple[str, str]] = []
|
||
for layer in layers:
|
||
if not isinstance(layer, dict):
|
||
normalized_layers.append(("", ""))
|
||
continue
|
||
role = str(layer.get("role") or "").strip().lower()
|
||
raw_path = str(layer.get("path") or "").strip()
|
||
normalized_path = (
|
||
os.path.normcase(os.path.abspath(raw_path))
|
||
if raw_path
|
||
else ""
|
||
)
|
||
normalized_layers.append((role, normalized_path))
|
||
if normalized_layers != [("user", expected_path)]:
|
||
descriptions = [
|
||
f"{role or 'unknown'}:{path or '<missing>'}"
|
||
for role, path in normalized_layers
|
||
]
|
||
raise GrokBuildError(
|
||
"Grok 实际配置含未受管的 requirements/system/MDM/project "
|
||
"或未知覆盖层（"
|
||
+ "、".join(descriptions or ["无可验证 user 层"])
|
||
+ "）；已阻止凭据注入"
|
||
)
|
||
permissions = (
|
||
inspection.get("permissions") if isinstance(inspection, dict) else None
|
||
)
|
||
if (
|
||
not isinstance(permissions, dict)
|
||
or permissions.get("managedSettingsExists") is not False
|
||
or permissions.get("managedSettingsActive") is not False
|
||
):
|
||
raise GrokBuildError(
|
||
"Grok 托管/MDM 设置状态无法证明为未启用；已阻止凭据注入"
|
||
)
|
||
|
||
try:
|
||
current_config = self.user_config_file.read_text(encoding="utf-8")
|
||
except OSError as exc:
|
||
raise GrokBuildError(
|
||
f"核验后无法重新读取 Grok config.toml；已阻止凭据注入：{exc}"
|
||
) from exc
|
||
if not hmac.compare_digest(
|
||
current_config.encode("utf-8"),
|
||
expected_user_config.encode("utf-8"),
|
||
):
|
||
raise GrokBuildError(
|
||
"Grok config.toml 在配置层核验期间发生变化；已阻止凭据注入"
|
||
)
|
||
if requirements_file.exists() or self.legacy_managed_config_file.exists():
|
||
raise GrokBuildError(
|
||
"Grok 高优先级配置层在核验期间出现；已阻止凭据注入"
|
||
)
|
||
|
||
def runtime_environment(
|
||
self,
|
||
include_model_key: bool = True,
|
||
include_mcp_secrets: bool | None = None,
|
||
workspace: str | os.PathLike[str] | None = None,
|
||
custom_model_only: bool = True,
|
||
) -> dict[str, str]:
|
||
env = dict(os.environ)
|
||
pending_secrets: dict[str, str] = {}
|
||
selected_workspace = Path(workspace or self.project_dir).resolve()
|
||
if not selected_workspace.is_dir():
|
||
raise GrokBuildError(f"工作目录不存在：{selected_workspace}")
|
||
env.pop(MODEL_API_KEY_ENV, None)
|
||
for variable in tuple(env):
|
||
if variable.startswith("WECOM_GROK_MCP_"):
|
||
env.pop(variable, None)
|
||
env["GROK_HOME"] = str(self.runtime_home)
|
||
env["PYTHONUTF8"] = "1"
|
||
env["GROK_DISABLE_AUTOUPDATER"] = "1"
|
||
if custom_model_only:
|
||
for variable in XAI_CREDENTIAL_ENV_VARS:
|
||
env.pop(variable, None)
|
||
for variable in tuple(env):
|
||
if (
|
||
variable in UNMANAGED_MODEL_ROUTE_ENV_VARS
|
||
or variable.startswith("GROK_AUTH_PROVIDER_")
|
||
):
|
||
env.pop(variable, None)
|
||
env.update(CUSTOM_MODEL_ENVIRONMENT)
|
||
no_xai_auth = (self.runtime_home / "no-xai-auth.json").resolve()
|
||
if no_xai_auth.exists():
|
||
raise GrokBuildError(
|
||
f"xAI 隔离认证路径必须不存在：{no_xai_auth}"
|
||
)
|
||
env["GROK_AUTH_PATH"] = str(no_xai_auth)
|
||
if include_mcp_secrets is None:
|
||
include_mcp_secrets = include_model_key
|
||
integration = self.load_integration_settings()
|
||
if not bool(integration.get("external_compatibility", False)):
|
||
for vendor in ("CURSOR", "CLAUDE", "CODEX"):
|
||
for surface in (
|
||
"SKILLS",
|
||
"RULES",
|
||
"AGENTS",
|
||
"MCPS",
|
||
"HOOKS",
|
||
"SESSIONS",
|
||
):
|
||
env[f"GROK_{vendor}_{surface}_ENABLED"] = "false"
|
||
try:
|
||
runtime_config = self.user_config_file.read_text(encoding="utf-8")
|
||
except (FileNotFoundError, OSError):
|
||
runtime_config = ""
|
||
managed_block_present = bool(self._managed_config_block())
|
||
if include_model_key or include_mcp_secrets:
|
||
try:
|
||
settings = self.load_ai_settings()
|
||
except GrokBuildError:
|
||
settings = {}
|
||
parsed_config = self._parse_toml_document(runtime_config)
|
||
else:
|
||
settings = {}
|
||
parsed_config = {}
|
||
models = parsed_config.get("models")
|
||
model_tables = parsed_config.get("model")
|
||
actual_model = (
|
||
model_tables.get(MODEL_PROFILE)
|
||
if isinstance(model_tables, dict)
|
||
else None
|
||
)
|
||
managed_model_exists = isinstance(actual_model, dict)
|
||
if include_model_key and not settings and managed_model_exists:
|
||
raise GrokBuildError(
|
||
"无法读取后台 Agent 自有模型配置；已阻止凭据回退，"
|
||
"请恢复 AI 设置并重新同步"
|
||
)
|
||
if include_model_key and settings:
|
||
profile = (
|
||
self.agent_model_profile(settings)
|
||
if custom_model_only
|
||
else self.model_profile(settings)
|
||
)
|
||
default_model = (
|
||
models.get("default") if isinstance(models, dict) else None
|
||
)
|
||
if (
|
||
profile.compatible
|
||
and actual_model == self._expected_model_table(profile)
|
||
and default_model == MODEL_PROFILE
|
||
and self._custom_model_routes_are_locked(
|
||
parsed_config,
|
||
runtime_config,
|
||
)
|
||
and managed_block_present
|
||
):
|
||
other_config = dict(parsed_config)
|
||
other_models = dict(model_tables) if isinstance(model_tables, dict) else {}
|
||
other_models.pop(MODEL_PROFILE, None)
|
||
other_config["model"] = other_models
|
||
if self._toml_references_environment(
|
||
other_config,
|
||
MODEL_API_KEY_ENV,
|
||
):
|
||
raise GrokBuildError(
|
||
f"Grok 配置在受管模型之外引用 {MODEL_API_KEY_ENV}；"
|
||
"已阻止密钥注入"
|
||
)
|
||
definition_violations = self._agent_definition_model_violations(
|
||
selected_workspace
|
||
)
|
||
if definition_violations:
|
||
raise GrokBuildError(
|
||
"检测到角色/Persona/Agent 文件指定了非受管模型；"
|
||
"已阻止启动，请改为 wecom-backend 或 inherit："
|
||
+ "、".join(definition_violations)
|
||
)
|
||
if custom_model_only:
|
||
key = self.agent_model_api_key(
|
||
settings,
|
||
profile=profile,
|
||
)
|
||
else:
|
||
key_name = (
|
||
"GROK_API_KEY"
|
||
if bool(settings.get("GROK_MODEL_ENABLED", False))
|
||
else "AI_API_KEY"
|
||
)
|
||
key = str(settings.get(key_name) or "").strip()
|
||
if key:
|
||
pending_secrets[MODEL_API_KEY_ENV] = key
|
||
# Do not allow an inherited xAI global credential to become
|
||
# a fallback for this third-party/custom provider.
|
||
env.pop("XAI_API_KEY", None)
|
||
else:
|
||
raise GrokBuildError(
|
||
"受管 Agent 自有模型缺少独立 API Key，请先在后台配置并重新同步"
|
||
)
|
||
elif managed_model_exists:
|
||
raise GrokBuildError(
|
||
"后台 Agent 自有模型与 Grok 受管配置不一致；已阻止凭据回退，"
|
||
"请先同步后台模型再启动任务"
|
||
)
|
||
elif custom_model_only:
|
||
raise GrokBuildError(
|
||
f"{profile.reason}；请先同步后台自有模型再启动 Agent"
|
||
)
|
||
if include_mcp_secrets and settings:
|
||
raw_servers = settings.get("AI_MCP_SERVERS")
|
||
integration = self.load_integration_settings()
|
||
actual_mcp_tables = parsed_config.get("mcp_servers")
|
||
actual_managed = (
|
||
{
|
||
str(name): value
|
||
for name, value in actual_mcp_tables.items()
|
||
if str(name).startswith(MANAGED_MCP_PREFIX)
|
||
}
|
||
if isinstance(actual_mcp_tables, dict)
|
||
else {}
|
||
)
|
||
actual_has_external_mcp = any(
|
||
name != CUSTOMER_SERVICE_MCP_NAME for name in actual_managed
|
||
)
|
||
expected_lines: list[str] = []
|
||
expected_servers: list[object] = []
|
||
if bool(integration.get("customer_service_tools", True)):
|
||
expected_servers.extend(self._customer_service_mcp_servers(settings))
|
||
if actual_has_external_mcp and isinstance(raw_servers, list):
|
||
expected_servers.extend(raw_servers)
|
||
if expected_servers:
|
||
expected_lines.extend(self._render_mcp_servers(expected_servers))
|
||
rendered_mcp = "\n".join(expected_lines).strip()
|
||
expected_mcp_config = (
|
||
self._parse_toml_document(rendered_mcp, "后台 MCP 配置")
|
||
if rendered_mcp
|
||
else {}
|
||
)
|
||
expected_mcp_tables = expected_mcp_config.get("mcp_servers")
|
||
expected_managed = (
|
||
{
|
||
str(name): value
|
||
for name, value in expected_mcp_tables.items()
|
||
if str(name).startswith(MANAGED_MCP_PREFIX)
|
||
}
|
||
if isinstance(expected_mcp_tables, dict)
|
||
else {}
|
||
)
|
||
if actual_managed and actual_managed != expected_managed:
|
||
raise GrokBuildError(
|
||
"后台 MCP 配置与 Grok 受管配置不一致；已阻止凭据注入，"
|
||
"请重新同步 MCP 配置"
|
||
)
|
||
if actual_managed and not managed_block_present:
|
||
raise GrokBuildError(
|
||
"Grok 受管 MCP 配置缺少唯一自动配置区块标记；"
|
||
"已阻止凭据注入，请重新同步 MCP 配置"
|
||
)
|
||
if actual_managed:
|
||
config_without_managed_mcp = dict(parsed_config)
|
||
config_without_managed_mcp["mcp_servers"] = {
|
||
str(name): value
|
||
for name, value in (
|
||
actual_mcp_tables.items()
|
||
if isinstance(actual_mcp_tables, dict)
|
||
else ()
|
||
)
|
||
if not str(name).startswith(MANAGED_MCP_PREFIX)
|
||
}
|
||
used: set[str] = (
|
||
{CUSTOMER_SERVICE_MCP_NAME}
|
||
if bool(integration.get("customer_service_tools", True))
|
||
else set()
|
||
)
|
||
for index, raw in enumerate(raw_servers, start=1):
|
||
if not isinstance(raw, dict) or raw.get("enabled") is False:
|
||
continue
|
||
base_name = self._safe_toml_identifier(
|
||
str(raw.get("name") or raw.get("id") or f"server-{index}")
|
||
)
|
||
base_name = f"{MANAGED_MCP_PREFIX}{base_name}"
|
||
name = base_name
|
||
suffix = 2
|
||
while name in used:
|
||
name = f"{base_name}-{suffix}"
|
||
suffix += 1
|
||
used.add(name)
|
||
if name not in actual_managed:
|
||
continue
|
||
for category, values in (
|
||
("env", raw.get("env")),
|
||
("header", raw.get("headers")),
|
||
):
|
||
if not isinstance(values, dict):
|
||
continue
|
||
for item_key, item_value in values.items():
|
||
rendered = str(item_value)
|
||
if re.fullmatch(
|
||
r"\$\{[A-Za-z_][A-Za-z0-9_]*\}",
|
||
rendered,
|
||
):
|
||
continue
|
||
variable = self._mcp_value_env_name(
|
||
name, category, item_key
|
||
)
|
||
if self._toml_references_environment(
|
||
config_without_managed_mcp,
|
||
variable,
|
||
):
|
||
raise GrokBuildError(
|
||
f"Grok 配置在受管 MCP 之外引用 {variable}；"
|
||
"已阻止密钥注入"
|
||
)
|
||
pending_secrets[variable] = rendered
|
||
if pending_secrets:
|
||
self._assert_effective_config_isolated(
|
||
env,
|
||
selected_workspace,
|
||
runtime_config,
|
||
)
|
||
env.update(pending_secrets)
|
||
return env
|
||
|
||
def locate_binary(self) -> Path | None:
|
||
integration = self.load_integration_settings()
|
||
candidates: list[Path] = []
|
||
override = str(os.environ.get("GROK_BUILD_BIN") or "").strip()
|
||
configured = str(integration.get("binary_path") or "").strip()
|
||
if override:
|
||
candidates.append(Path(override).expanduser())
|
||
if configured:
|
||
candidates.append(Path(configured).expanduser())
|
||
candidates.append(self.binary_path)
|
||
seen: set[str] = set()
|
||
for candidate in candidates:
|
||
normalized = os.path.normcase(os.path.abspath(str(candidate)))
|
||
if normalized in seen:
|
||
continue
|
||
seen.add(normalized)
|
||
if candidate.is_file():
|
||
return candidate.resolve()
|
||
return None
|
||
|
||
@staticmethod
|
||
def _file_sha256(path: Path) -> str:
|
||
digest = hashlib.sha256()
|
||
with path.open("rb") as handle:
|
||
for block in iter(lambda: handle.read(1024 * 1024), b""):
|
||
digest.update(block)
|
||
return digest.hexdigest()
|
||
|
||
def validate_binary(self, binary: Path) -> None:
|
||
target = binary.resolve()
|
||
stat = target.stat()
|
||
fingerprint = (str(target), stat.st_size, stat.st_mtime_ns)
|
||
if self._validated_binary_fingerprint == fingerprint:
|
||
return
|
||
if stat.st_size < 1024 * 1024:
|
||
raise GrokBuildError(f"Grok Build 运行时文件异常：{target}")
|
||
|
||
trusted_install_record = False
|
||
if target == self.binary_path.resolve() and self.install_state_file.is_file():
|
||
try:
|
||
state = json.loads(
|
||
self.install_state_file.read_text(encoding="utf-8")
|
||
)
|
||
expected = str(state.get("sha256") or "").strip().lower()
|
||
source = str(state.get("source") or "").strip()
|
||
publisher = str(state.get("publisher") or "").strip()
|
||
except (OSError, ValueError, TypeError) as exc:
|
||
raise GrokBuildError(f"Grok Build 安装记录损坏：{exc}") from exc
|
||
if expected:
|
||
actual = self._file_sha256(target)
|
||
if not hmac.compare_digest(actual, expected):
|
||
raise GrokBuildError("Grok Build 运行时 SHA-256 与安装记录不一致")
|
||
trusted_install_record = (
|
||
source
|
||
in {
|
||
"x.ai/cli",
|
||
OFFICIAL_BASE_URL,
|
||
OFFICIAL_FALLBACK_URL,
|
||
}
|
||
and bool(publisher)
|
||
and re.search(
|
||
r"(?:^|,\s*)CN=X\.AI LLC(?:,|$)",
|
||
publisher.upper(),
|
||
)
|
||
is not None
|
||
)
|
||
# The installer performs Authenticode verification before recording the
|
||
# SHA-256. Re-hashing that exact managed binary is sufficient on later
|
||
# launches and avoids a slow certificate-chain lookup at every startup.
|
||
# Explicit external binaries and legacy records are always re-verified.
|
||
if os.name == "nt" and not trusted_install_record:
|
||
self._verify_windows_signature(target)
|
||
self._validated_binary_fingerprint = fingerprint
|
||
|
||
def version(self, binary: Path | None = None, timeout: float = 8.0) -> str:
|
||
target = binary or self.locate_binary()
|
||
if target is None:
|
||
return ""
|
||
self.validate_binary(target)
|
||
try:
|
||
completed = subprocess.run(
|
||
[str(target), "--version"],
|
||
cwd=str(self.project_dir),
|
||
env=self.runtime_environment(include_model_key=False),
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.STDOUT,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
timeout=timeout,
|
||
check=False,
|
||
)
|
||
except (OSError, subprocess.SubprocessError):
|
||
return ""
|
||
first_line = (completed.stdout or "").strip().splitlines()
|
||
return first_line[0].strip() if first_line else ""
|
||
|
||
@staticmethod
|
||
def _dify_adapter_instance_live(sync: ModelSyncResult) -> bool:
|
||
base_url = str(sync.effective_base_url or "").strip()
|
||
instance_id = str(sync.adapter_instance_id or "").strip()
|
||
if not base_url or not instance_id:
|
||
return False
|
||
try:
|
||
parsed = urlsplit(base_url)
|
||
port = parsed.port
|
||
except ValueError:
|
||
return False
|
||
if (
|
||
parsed.scheme != "http"
|
||
or parsed.hostname != "127.0.0.1"
|
||
or not port
|
||
):
|
||
return False
|
||
health_url = urlunsplit(
|
||
(parsed.scheme, parsed.netloc, "/health", "", "")
|
||
)
|
||
request = urllib.request.Request(
|
||
health_url,
|
||
headers={"Accept": "application/json", "User-Agent": USER_AGENT},
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=0.75) as response:
|
||
if (
|
||
int(response.getcode()) != 200
|
||
or str(response.headers.get("X-Grok-Dify-Adapter") or "")
|
||
!= "1"
|
||
):
|
||
return False
|
||
payload = json.loads(response.read(4097).decode("utf-8"))
|
||
except (OSError, ValueError, urllib.error.URLError):
|
||
return False
|
||
return bool(
|
||
isinstance(payload, dict)
|
||
and payload.get("ok") is True
|
||
and payload.get("adapter") == "dify"
|
||
and hmac.compare_digest(
|
||
str(payload.get("instance_id") or ""),
|
||
instance_id,
|
||
)
|
||
)
|
||
|
||
def status(self) -> RuntimeStatus:
|
||
binary = self.locate_binary()
|
||
sync = self.read_sync_result()
|
||
source_backend = sync.source_api_backend or sync.api_backend
|
||
adapter_live = (
|
||
self._dify_adapter_instance_live(sync)
|
||
if source_backend == "dify"
|
||
else False
|
||
)
|
||
return RuntimeStatus(
|
||
installed=binary is not None,
|
||
binary_path=str(binary or self.binary_path),
|
||
version=self.version(binary) if binary else "",
|
||
# Kept for backward-compatible status JSON. xAI authentication is
|
||
# disabled and is never considered a readiness signal.
|
||
authenticated=False,
|
||
runtime_home=str(self.runtime_home),
|
||
model_configured=sync.configured,
|
||
model_compatible=sync.compatible,
|
||
model_name=sync.model,
|
||
model_message=sync.message,
|
||
model_api_backend=source_backend,
|
||
model_effective_base_url=sync.effective_base_url,
|
||
adapter_live=adapter_live,
|
||
warnings=tuple(self.migration_warnings),
|
||
)
|
||
|
||
@staticmethod
|
||
def _request(url: str, timeout: float = 30.0):
|
||
request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
|
||
return urllib.request.urlopen(request, timeout=timeout)
|
||
|
||
def latest_version(self, channel: str | None = None) -> str:
|
||
selected = str(
|
||
channel or self.load_integration_settings().get("runtime_channel") or "stable"
|
||
).strip()
|
||
if selected not in {"stable", "alpha", "enterprise"}:
|
||
raise GrokBuildError(f"不支持的更新通道：{selected}")
|
||
errors: list[str] = []
|
||
for base_url in (OFFICIAL_BASE_URL, OFFICIAL_FALLBACK_URL):
|
||
try:
|
||
with self._request(f"{base_url}/{selected}", timeout=20) as response:
|
||
version = response.read().decode("utf-8", "replace").strip()
|
||
except (OSError, urllib.error.URLError) as exc:
|
||
errors.append(str(exc))
|
||
continue
|
||
if VERSION_PATTERN.fullmatch(version):
|
||
return version
|
||
errors.append(f"{base_url} 返回了无效版本号")
|
||
raise GrokBuildError("无法获取 Grok Build 最新版本：" + "；".join(errors))
|
||
|
||
@staticmethod
|
||
def _platform_tag() -> str:
|
||
machine = platform.machine().lower()
|
||
if machine in {"amd64", "x86_64", "x64"}:
|
||
architecture = "x86_64"
|
||
elif machine in {"arm64", "aarch64"}:
|
||
architecture = "aarch64"
|
||
else:
|
||
raise GrokBuildError(f"不支持的处理器架构：{platform.machine()}")
|
||
if os.name == "nt":
|
||
return f"windows-{architecture}"
|
||
if sys.platform == "darwin":
|
||
return f"darwin-{architecture}"
|
||
if sys.platform.startswith("linux"):
|
||
return f"linux-{architecture}"
|
||
raise GrokBuildError(f"不支持的操作系统：{sys.platform}")
|
||
|
||
@staticmethod
|
||
def _verify_windows_signature(path: Path) -> str:
|
||
if os.name != "nt":
|
||
return ""
|
||
system_root = Path(os.environ.get("SystemRoot") or r"C:\Windows")
|
||
powershell = (
|
||
system_root
|
||
/ "System32"
|
||
/ "WindowsPowerShell"
|
||
/ "v1.0"
|
||
/ "powershell.exe"
|
||
)
|
||
if not powershell.is_file():
|
||
discovered = shutil.which("powershell") or shutil.which("pwsh")
|
||
if not discovered:
|
||
raise GrokBuildError("无法校验官方运行时数字签名：未找到 PowerShell")
|
||
powershell = Path(discovered)
|
||
script = (
|
||
"& { param([string]$p) "
|
||
"$s = Get-AuthenticodeSignature -LiteralPath $p; "
|
||
"[ordered]@{status=[string]$s.Status;"
|
||
"subject=[string]$s.SignerCertificate.Subject;"
|
||
"simple_name=[string]$s.SignerCertificate.GetNameInfo("
|
||
"[System.Security.Cryptography.X509Certificates.X509NameType]::SimpleName,"
|
||
"$false)} | ConvertTo-Json -Compress }"
|
||
)
|
||
try:
|
||
completed = subprocess.run(
|
||
[
|
||
str(powershell),
|
||
"-NoLogo",
|
||
"-NoProfile",
|
||
"-NonInteractive",
|
||
"-ExecutionPolicy",
|
||
"Bypass",
|
||
"-Command",
|
||
script,
|
||
str(path),
|
||
],
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
timeout=45,
|
||
check=False,
|
||
)
|
||
except (OSError, subprocess.SubprocessError) as exc:
|
||
raise GrokBuildError(f"无法校验官方运行时数字签名：{exc}") from exc
|
||
try:
|
||
result = json.loads((completed.stdout or "").strip())
|
||
except (ValueError, TypeError) as exc:
|
||
detail = (completed.stderr or completed.stdout or "没有签名校验输出").strip()
|
||
raise GrokBuildError(f"无法解析数字签名结果：{detail}") from exc
|
||
status = str(result.get("status") or "")
|
||
subject = str(result.get("subject") or "")
|
||
simple_name = str(result.get("simple_name") or "").strip().casefold()
|
||
if completed.returncode != 0 or status.lower() != "valid":
|
||
raise GrokBuildError(f"Grok Build 数字签名无效：{status or 'Unknown'}")
|
||
if simple_name not in {"x.ai llc", "xai llc"}:
|
||
raise GrokBuildError(f"Grok Build 发布者不受信任：{subject or 'Unknown'}")
|
||
return subject
|
||
|
||
def install_official_release(
|
||
self,
|
||
version: str | None = None,
|
||
progress: Callable[[int, int], None] | None = None,
|
||
) -> RuntimeStatus:
|
||
resolved_version = (version or self.latest_version()).strip()
|
||
if not VERSION_PATTERN.fullmatch(resolved_version):
|
||
raise GrokBuildError(f"无效版本号：{resolved_version}")
|
||
tag = self._platform_tag()
|
||
extension = ".exe" if os.name == "nt" else ""
|
||
artifact = f"grok-{resolved_version}-{tag}{extension}"
|
||
self.bin_dir.mkdir(parents=True, exist_ok=True)
|
||
downloads = self.binary_store_home / "downloads"
|
||
downloads.mkdir(parents=True, exist_ok=True)
|
||
temporary = downloads / f"{artifact}.{os.getpid()}.part"
|
||
errors: list[str] = []
|
||
downloaded = False
|
||
download_source = ""
|
||
for base_url in (OFFICIAL_BASE_URL, OFFICIAL_FALLBACK_URL):
|
||
url = f"{base_url}/{artifact}"
|
||
try:
|
||
with self._request(url, timeout=300) as response, temporary.open("wb") as target:
|
||
try:
|
||
total = int(response.headers.get("Content-Length") or 0)
|
||
except (TypeError, ValueError):
|
||
total = 0
|
||
received = 0
|
||
while True:
|
||
block = response.read(1024 * 256)
|
||
if not block:
|
||
break
|
||
target.write(block)
|
||
received += len(block)
|
||
if progress:
|
||
progress(received, total)
|
||
downloaded = True
|
||
download_source = base_url
|
||
break
|
||
except (OSError, urllib.error.URLError) as exc:
|
||
errors.append(f"{url}: {exc}")
|
||
try:
|
||
temporary.unlink()
|
||
except FileNotFoundError:
|
||
pass
|
||
if not downloaded:
|
||
raise GrokBuildError("官方运行时下载失败：" + "；".join(errors))
|
||
size = temporary.stat().st_size
|
||
if size < 1024 * 1024:
|
||
temporary.unlink(missing_ok=True)
|
||
raise GrokBuildError(f"下载文件异常，仅 {size} 字节")
|
||
if os.name == "nt":
|
||
with temporary.open("rb") as handle:
|
||
if handle.read(2) != b"MZ":
|
||
temporary.unlink(missing_ok=True)
|
||
raise GrokBuildError("下载文件不是有效的 Windows 可执行文件")
|
||
publisher = self._verify_windows_signature(temporary)
|
||
else:
|
||
temporary.chmod(0o755)
|
||
publisher = ""
|
||
digest = hashlib.sha256()
|
||
with temporary.open("rb") as handle:
|
||
for block in iter(lambda: handle.read(1024 * 1024), b""):
|
||
digest.update(block)
|
||
sha256 = digest.hexdigest()
|
||
os.replace(temporary, self.binary_path)
|
||
if os.name != "nt":
|
||
self.binary_path.chmod(0o755)
|
||
try:
|
||
if self.agent_alias_path.exists():
|
||
self.agent_alias_path.unlink()
|
||
os.link(self.binary_path, self.agent_alias_path)
|
||
except OSError:
|
||
shutil.copy2(self.binary_path, self.agent_alias_path)
|
||
self._atomic_write(
|
||
self.install_state_file,
|
||
json.dumps(
|
||
{
|
||
"version": resolved_version,
|
||
"platform": tag,
|
||
"installed_at": _utc_timestamp(),
|
||
"binary_path": str(self.binary_path),
|
||
"source": download_source,
|
||
"sha256": sha256,
|
||
"publisher": publisher,
|
||
},
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
)
|
||
+ "\n",
|
||
)
|
||
return self.status()
|
||
|
||
def require_binary(self) -> Path:
|
||
binary = self.locate_binary()
|
||
if binary is None:
|
||
raise GrokBuildError("尚未安装 Grok Build，请先执行安装")
|
||
self.validate_binary(binary)
|
||
return binary
|
||
|
||
def build_headless_args(
|
||
self,
|
||
prompt: str,
|
||
*,
|
||
workspace: str | os.PathLike[str] | None = None,
|
||
model: str = "",
|
||
effort: str = "high",
|
||
max_turns: int = 50,
|
||
auto_approve: bool = False,
|
||
read_only: bool = False,
|
||
continue_session: bool = False,
|
||
new_session_id: str = "",
|
||
resume_session: str = "",
|
||
sandbox: str = "",
|
||
allowed_tools: str = "",
|
||
disallowed_tools: str = "",
|
||
rules: str = "",
|
||
) -> list[str]:
|
||
if not prompt.strip():
|
||
raise GrokBuildError("任务内容不能为空")
|
||
selected_workspace = Path(workspace or self.project_dir).resolve()
|
||
if not selected_workspace.is_dir():
|
||
raise GrokBuildError(f"工作目录不存在：{selected_workspace}")
|
||
args = [
|
||
"-p",
|
||
prompt,
|
||
"--cwd",
|
||
str(selected_workspace),
|
||
"--output-format",
|
||
"streaming-json",
|
||
"--max-turns",
|
||
str(min(10000, max(1, int(max_turns)))),
|
||
"--no-auto-update",
|
||
]
|
||
if model.strip():
|
||
args.extend(["--model", model.strip()])
|
||
if effort.strip():
|
||
args.extend(["--reasoning-effort", effort.strip()])
|
||
if auto_approve:
|
||
args.append("--yolo")
|
||
if read_only:
|
||
allowed_tools = "read_file,grep,list_dir,web_search,web_fetch"
|
||
mandatory_denied = {"search_tool", "use_tool", "Agent"}
|
||
mandatory_denied.update(
|
||
item.strip() for item in disallowed_tools.split(",") if item.strip()
|
||
)
|
||
disallowed_tools = ",".join(sorted(mandatory_denied))
|
||
args.append("--no-subagents")
|
||
if self._agent_uses_dify_source():
|
||
denied = [
|
||
item.strip()
|
||
for item in disallowed_tools.split(",")
|
||
if item.strip()
|
||
]
|
||
if "web_search" not in denied:
|
||
denied.append("web_search")
|
||
disallowed_tools = ",".join(denied)
|
||
if allowed_tools.strip():
|
||
args.extend(["--tools", allowed_tools.strip()])
|
||
if disallowed_tools.strip():
|
||
args.extend(["--disallowed-tools", disallowed_tools.strip()])
|
||
if new_session_id.strip() and (
|
||
resume_session.strip() or continue_session
|
||
):
|
||
raise GrokBuildError(
|
||
"新会话、恢复指定会话和继续最近会话不能同时启用"
|
||
)
|
||
if new_session_id.strip():
|
||
raw_session_id = new_session_id.strip()
|
||
try:
|
||
normalized_session_id = str(uuid.UUID(raw_session_id))
|
||
except (ValueError, AttributeError) as exc:
|
||
raise GrokBuildError("新会话 ID 必须是有效 UUID") from exc
|
||
if raw_session_id != normalized_session_id:
|
||
raise GrokBuildError("新会话 ID 必须是规范的小写 UUID")
|
||
args.extend(["--session-id", normalized_session_id])
|
||
elif resume_session.strip():
|
||
args.extend(["--resume", resume_session.strip()])
|
||
elif continue_session:
|
||
args.append("--continue")
|
||
if sandbox.strip():
|
||
args.extend(["--sandbox", sandbox.strip()])
|
||
if rules.strip():
|
||
args.extend(["--rules", rules.strip()])
|
||
return args
|
||
|
||
def build_acp_args(
|
||
self,
|
||
*,
|
||
model: str = "",
|
||
yolo: bool = False,
|
||
sandbox: str = "",
|
||
) -> list[str]:
|
||
args: list[str] = ["--no-auto-update"]
|
||
if self._agent_uses_dify_source():
|
||
args.append("--disable-web-search")
|
||
if sandbox.strip():
|
||
args.extend(["--sandbox", sandbox.strip()])
|
||
args.extend(["agent", "--no-leader"])
|
||
if model.strip():
|
||
args.extend(["--model", model.strip()])
|
||
if yolo:
|
||
args.append("--always-approve")
|
||
args.append("stdio")
|
||
return args
|
||
|
||
def run_capture(
|
||
self,
|
||
args: Sequence[str],
|
||
*,
|
||
workspace: str | os.PathLike[str] | None = None,
|
||
timeout: float = 60.0,
|
||
include_managed_secrets: bool = False,
|
||
) -> subprocess.CompletedProcess[str]:
|
||
binary = self.require_binary()
|
||
if include_managed_secrets:
|
||
self.prepare_agent_configuration()
|
||
selected_workspace = Path(workspace or self.project_dir).resolve()
|
||
return subprocess.run(
|
||
[str(binary), *args],
|
||
cwd=str(selected_workspace),
|
||
env=self.runtime_environment(
|
||
include_model_key=include_managed_secrets,
|
||
include_mcp_secrets=include_managed_secrets,
|
||
workspace=selected_workspace,
|
||
custom_model_only=True,
|
||
),
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.STDOUT,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
timeout=timeout,
|
||
check=False,
|
||
)
|
||
|
||
def verify_read_only_environment(
|
||
self,
|
||
workspace: str | os.PathLike[str] | None = None,
|
||
) -> None:
|
||
"""Refuse read-only mode when native executable extensions are active.
|
||
|
||
Upstream starts configured MCP servers, plugin hooks, and LSP servers
|
||
outside the built-in tool allow-list. A metadata-only inspect runs
|
||
first without managed secrets; the model process is not started unless
|
||
the effective configuration is free of those executable surfaces.
|
||
"""
|
||
selected_workspace = Path(workspace or self.project_dir).resolve()
|
||
if not selected_workspace.is_dir():
|
||
raise GrokBuildError(f"工作目录不存在：{selected_workspace}")
|
||
completed = self.run_capture(
|
||
["--no-auto-update", "inspect", "--json"],
|
||
workspace=selected_workspace,
|
||
timeout=30,
|
||
include_managed_secrets=False,
|
||
)
|
||
if completed.returncode != 0:
|
||
raise GrokBuildError(
|
||
"无法验证只读环境，Grok inspect 退出码为 "
|
||
f"{completed.returncode}：{(completed.stdout or '').strip()}"
|
||
)
|
||
try:
|
||
inspection = json.loads((completed.stdout or "").lstrip("\ufeff"))
|
||
except json.JSONDecodeError as exc:
|
||
raise GrokBuildError(
|
||
"无法验证只读环境：Grok inspect 未返回有效 JSON"
|
||
) from exc
|
||
if not isinstance(inspection, dict):
|
||
raise GrokBuildError("无法验证只读环境：Grok inspect 返回结构无效")
|
||
|
||
risks: list[str] = []
|
||
hooks = inspection.get("hooks", [])
|
||
if not isinstance(hooks, list):
|
||
raise GrokBuildError("无法验证只读环境：hooks 检查结果结构无效")
|
||
for hook in hooks:
|
||
if not isinstance(hook, dict):
|
||
risks.append("Hook（未知来源）")
|
||
continue
|
||
source = hook.get("source")
|
||
plugin_name = (
|
||
str(source.get("plugin_name") or "").strip()
|
||
if isinstance(source, dict)
|
||
else ""
|
||
)
|
||
target = str(hook.get("target") or "").strip()
|
||
risks.append(f"Hook {plugin_name or target or '未知来源'}")
|
||
|
||
plugins = inspection.get("plugins", [])
|
||
if not isinstance(plugins, list):
|
||
raise GrokBuildError("无法验证只读环境：plugins 检查结果结构无效")
|
||
for plugin in plugins:
|
||
if not isinstance(plugin, dict) or plugin.get("enabled") is not True:
|
||
continue
|
||
provides = plugin.get("provides")
|
||
executable = not isinstance(provides, dict) or bool(
|
||
provides.get("hooks")
|
||
or provides.get("mcpServers")
|
||
or provides.get("lspServers")
|
||
)
|
||
if executable:
|
||
risks.append(
|
||
f"插件 {str(plugin.get('name') or '未知名称').strip()}"
|
||
)
|
||
|
||
mcp_servers = inspection.get("mcpServers", [])
|
||
if not isinstance(mcp_servers, list):
|
||
raise GrokBuildError("无法验证只读环境：mcpServers 检查结果结构无效")
|
||
for server in mcp_servers:
|
||
if not isinstance(server, dict) or server.get("disabled") is not True:
|
||
name = (
|
||
str(server.get("name") or "未知名称").strip()
|
||
if isinstance(server, dict)
|
||
else "未知名称"
|
||
)
|
||
risks.append(f"MCP {name}")
|
||
|
||
lsp_servers = inspection.get("lspServers", [])
|
||
if not isinstance(lsp_servers, list):
|
||
raise GrokBuildError("无法验证只读环境：lspServers 检查结果结构无效")
|
||
for server in lsp_servers:
|
||
if not isinstance(server, dict) or server.get("disabled") is not True:
|
||
name = (
|
||
str(server.get("name") or "未知名称").strip()
|
||
if isinstance(server, dict)
|
||
else "未知名称"
|
||
)
|
||
risks.append(f"LSP {name}")
|
||
|
||
if risks:
|
||
summary = "、".join(dict.fromkeys(risks))
|
||
raise GrokBuildError(
|
||
"只读审查拒绝启动：当前 Grok 配置含可执行扩展（"
|
||
f"{summary}）。这些扩展可能在模型请求前启动并继承凭据；"
|
||
"请先在完整 TUI/配置中禁用后重试，或取消“只读审查”使用逐项审批。"
|
||
)
|
||
|
||
def launch_console(
|
||
self,
|
||
args: Sequence[str] = (),
|
||
*,
|
||
workspace: str | os.PathLike[str] | None = None,
|
||
include_model_key: bool = True,
|
||
include_mcp_secrets: bool = True,
|
||
custom_model_only: bool = True,
|
||
) -> subprocess.Popen:
|
||
binary = self.require_binary()
|
||
if include_model_key:
|
||
self.prepare_agent_configuration()
|
||
selected_workspace = Path(workspace or self.project_dir).resolve()
|
||
if not selected_workspace.is_dir():
|
||
raise GrokBuildError(f"工作目录不存在：{selected_workspace}")
|
||
kwargs: dict[str, object] = {
|
||
"cwd": str(selected_workspace),
|
||
"env": self.runtime_environment(
|
||
include_model_key=include_model_key,
|
||
include_mcp_secrets=include_mcp_secrets,
|
||
workspace=selected_workspace,
|
||
custom_model_only=custom_model_only,
|
||
),
|
||
}
|
||
if os.name == "nt":
|
||
kwargs["creationflags"] = subprocess.CREATE_NEW_CONSOLE
|
||
forwarded = list(args)
|
||
if "--no-auto-update" not in forwarded:
|
||
forwarded.insert(0, "--no-auto-update")
|
||
return subprocess.Popen([str(binary), *forwarded], **kwargs)
|
||
|
||
def open_tui(
|
||
self,
|
||
*,
|
||
workspace: str | os.PathLike[str] | None = None,
|
||
initial_prompt: str = "",
|
||
model: str = "",
|
||
) -> subprocess.Popen:
|
||
selected_workspace = Path(workspace or self.project_dir).resolve()
|
||
requested_model = model.strip()
|
||
if requested_model and requested_model != MODEL_PROFILE:
|
||
raise GrokBuildError(
|
||
f"本项目的 Grok Agent 只允许使用后台受管模型 {MODEL_PROFILE}"
|
||
)
|
||
args = [
|
||
"--cwd",
|
||
str(selected_workspace),
|
||
"--model",
|
||
MODEL_PROFILE,
|
||
]
|
||
if self._agent_uses_dify_source():
|
||
args.insert(0, "--disable-web-search")
|
||
if initial_prompt.strip():
|
||
args.append(initial_prompt.strip())
|
||
return self.launch_console(
|
||
args,
|
||
workspace=selected_workspace,
|
||
include_model_key=True,
|
||
include_mcp_secrets=True,
|
||
custom_model_only=True,
|
||
)
|
||
|
||
def start_acp(
|
||
self,
|
||
*,
|
||
workspace: str | os.PathLike[str] | None = None,
|
||
model: str = "",
|
||
yolo: bool = False,
|
||
sandbox: str = "",
|
||
) -> subprocess.Popen:
|
||
binary = self.require_binary()
|
||
self.prepare_agent_configuration()
|
||
selected_workspace = Path(workspace or self.project_dir).resolve()
|
||
requested_model = model.strip()
|
||
if requested_model and requested_model != MODEL_PROFILE:
|
||
raise GrokBuildError(
|
||
f"本项目的 Grok Agent 只允许使用后台受管模型 {MODEL_PROFILE}"
|
||
)
|
||
return subprocess.Popen(
|
||
[
|
||
str(binary),
|
||
*self.build_acp_args(
|
||
model=MODEL_PROFILE,
|
||
yolo=yolo,
|
||
sandbox=sandbox,
|
||
),
|
||
],
|
||
cwd=str(selected_workspace),
|
||
env=self.runtime_environment(
|
||
include_model_key=True,
|
||
include_mcp_secrets=True,
|
||
workspace=selected_workspace,
|
||
custom_model_only=True,
|
||
),
|
||
stdin=subprocess.PIPE,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
)
|
||
|
||
|
||
def _status_dict(status: RuntimeStatus) -> dict:
|
||
return asdict(status)
|
||
|
||
|
||
def normalize_passthrough_args(args: Iterable[str]) -> list[str]:
|
||
"""Accept the conventional ``--`` separator without forwarding it to Grok."""
|
||
forwarded = list(args)
|
||
if forwarded[:1] == ["--"]:
|
||
forwarded = forwarded[1:]
|
||
if not forwarded:
|
||
raise GrokBuildError("exec 后必须提供 Grok 命令参数")
|
||
return forwarded
|
||
|
||
|
||
def resolve_passthrough_workspace(
|
||
args: Sequence[str],
|
||
process_workspace: str | os.PathLike[str],
|
||
) -> Path:
|
||
"""Resolve Grok's effective ``--cwd`` for configuration-layer checks."""
|
||
base = Path(process_workspace).expanduser().resolve()
|
||
selected = base
|
||
index = 0
|
||
while index < len(args):
|
||
token = str(args[index]).strip()
|
||
if token == "--cwd":
|
||
if index + 1 >= len(args) or not str(args[index + 1]).strip():
|
||
raise GrokBuildError("Grok --cwd 缺少目录参数")
|
||
raw = Path(str(args[index + 1]).strip()).expanduser()
|
||
selected = (base / raw).resolve() if not raw.is_absolute() else raw.resolve()
|
||
index += 2
|
||
continue
|
||
if token.startswith("--cwd="):
|
||
raw_value = token.split("=", 1)[1].strip()
|
||
if not raw_value:
|
||
raise GrokBuildError("Grok --cwd 缺少目录参数")
|
||
raw = Path(raw_value).expanduser()
|
||
selected = (base / raw).resolve() if not raw.is_absolute() else raw.resolve()
|
||
index += 1
|
||
if not selected.is_dir():
|
||
raise GrokBuildError(f"工作目录不存在：{selected}")
|
||
return selected
|
||
|
||
|
||
def classify_passthrough_args(args: Sequence[str]) -> str:
|
||
"""Classify upstream CLI passthrough without guessing about new commands.
|
||
|
||
``wrap`` is discovered in a complete first pass so earlier agent flags
|
||
cannot cause arbitrary wrapped child processes to receive managed secrets.
|
||
"""
|
||
safe_commands = {
|
||
"completions",
|
||
"doctor",
|
||
"export",
|
||
"help",
|
||
"inspect",
|
||
"leader",
|
||
"login",
|
||
"logout",
|
||
"mcp",
|
||
"memory",
|
||
"models",
|
||
"plugin",
|
||
"sessions",
|
||
"setup",
|
||
"trace",
|
||
"update",
|
||
"version",
|
||
"worktree",
|
||
}
|
||
agent_commands = {"agent", "dashboard"}
|
||
value_options = {
|
||
"--agent",
|
||
"--agents",
|
||
"--allow",
|
||
"--cwd",
|
||
"--debug-file",
|
||
"--deny",
|
||
"--disallowed-tools",
|
||
"--json-schema",
|
||
"--leader-socket",
|
||
"-m",
|
||
"--model",
|
||
"--max-turns",
|
||
"--output-format",
|
||
"--permission-mode",
|
||
"--reasoning-effort",
|
||
"--rules",
|
||
"-s",
|
||
"--session-id",
|
||
"--sandbox",
|
||
"--system-prompt-override",
|
||
"--tools",
|
||
"--worktree-ref",
|
||
}
|
||
required_agent_values = {
|
||
"-p",
|
||
"--single",
|
||
"--prompt-file",
|
||
"--prompt-json",
|
||
}
|
||
optional_agent_values = {"-r", "--resume", "-w", "--worktree"}
|
||
boolean_agent_flags = {"-c", "--continue"}
|
||
if not args:
|
||
return "unknown"
|
||
|
||
# Security priority pass: find a real top-level wrap token while skipping
|
||
# values belonging to options. False positives for optional resume or
|
||
# worktree values are intentionally treated as wrap (no secrets).
|
||
index = 0
|
||
value_taking = value_options | required_agent_values
|
||
while index < len(args):
|
||
token = str(args[index]).strip()
|
||
lowered = token.lower()
|
||
if (
|
||
lowered in {"--plugin-dir", "--agent-profile"}
|
||
or lowered.startswith("--plugin-dir=")
|
||
or lowered.startswith("--agent-profile=")
|
||
):
|
||
return "extension"
|
||
if lowered in value_taking:
|
||
index += 2
|
||
continue
|
||
if any(
|
||
lowered.startswith(option + "=")
|
||
for option in value_taking
|
||
if option.startswith("--")
|
||
) or lowered.startswith("-p="):
|
||
index += 1
|
||
continue
|
||
if lowered == "wrap":
|
||
return "wrap"
|
||
index += 1
|
||
|
||
index = 0
|
||
agent_requested = False
|
||
while index < len(args):
|
||
token = str(args[index]).strip()
|
||
lowered = token.lower()
|
||
if lowered in {"--help", "-h", "--version", "-v"}:
|
||
return "safe"
|
||
if lowered in required_agent_values:
|
||
agent_requested = True
|
||
index += 2
|
||
continue
|
||
if any(
|
||
lowered.startswith(flag + "=")
|
||
for flag in required_agent_values
|
||
if flag.startswith("--")
|
||
) or lowered.startswith("-p="):
|
||
agent_requested = True
|
||
index += 1
|
||
continue
|
||
if lowered in optional_agent_values:
|
||
agent_requested = True
|
||
if (
|
||
index + 1 < len(args)
|
||
and not str(args[index + 1]).startswith("-")
|
||
and str(args[index + 1]).strip().lower()
|
||
not in safe_commands | agent_commands | {"wrap"}
|
||
):
|
||
index += 2
|
||
else:
|
||
index += 1
|
||
continue
|
||
if any(
|
||
lowered.startswith(flag + "=")
|
||
for flag in optional_agent_values
|
||
if flag.startswith("--")
|
||
):
|
||
agent_requested = True
|
||
index += 1
|
||
continue
|
||
if lowered in boolean_agent_flags:
|
||
agent_requested = True
|
||
index += 1
|
||
continue
|
||
if lowered in value_options:
|
||
index += 2
|
||
continue
|
||
if any(
|
||
lowered.startswith(option + "=")
|
||
for option in value_options
|
||
if option.startswith("--")
|
||
):
|
||
index += 1
|
||
continue
|
||
if lowered.startswith("-"):
|
||
index += 1
|
||
continue
|
||
if lowered == "wrap":
|
||
return "wrap"
|
||
if lowered in safe_commands:
|
||
return "safe"
|
||
if lowered in agent_commands:
|
||
return "agent"
|
||
return "unknown"
|
||
return "agent" if agent_requested else "unknown"
|
||
|
||
|
||
def enforce_passthrough_agent_model(args: Sequence[str]) -> list[str]:
|
||
"""Pin passthrough Agent invocations to the backend-managed profile."""
|
||
forwarded = list(args)
|
||
found_model = False
|
||
index = 0
|
||
while index < len(forwarded):
|
||
token = str(forwarded[index]).strip()
|
||
lowered = token.lower()
|
||
if lowered in {"-m", "--model"}:
|
||
if index + 1 >= len(forwarded):
|
||
raise GrokBuildError(f"{token} 缺少模型参数")
|
||
requested = str(forwarded[index + 1]).strip()
|
||
if requested != MODEL_PROFILE:
|
||
raise GrokBuildError(
|
||
f"Grok Agent 只允许使用后台受管模型 {MODEL_PROFILE}"
|
||
)
|
||
found_model = True
|
||
index += 2
|
||
continue
|
||
if lowered.startswith("--model="):
|
||
requested = token.split("=", 1)[1].strip()
|
||
if requested != MODEL_PROFILE:
|
||
raise GrokBuildError(
|
||
f"Grok Agent 只允许使用后台受管模型 {MODEL_PROFILE}"
|
||
)
|
||
found_model = True
|
||
index += 1
|
||
if not found_model:
|
||
forwarded[0:0] = ["--model", MODEL_PROFILE]
|
||
return forwarded
|
||
|
||
|
||
def reject_passthrough_model_bypasses(args: Sequence[str]) -> None:
|
||
"""Reject high-priority CLI surfaces that can bypass managed routing."""
|
||
profile_options = {"--agent", "--agents", "--agent-profile", "--plugin-dir"}
|
||
model_options = {
|
||
"-m",
|
||
"--model",
|
||
"--web-search-model",
|
||
"--session-summary-model",
|
||
"--image-description-model",
|
||
"--prompt-suggestions-model",
|
||
}
|
||
index = 0
|
||
while index < len(args):
|
||
token = str(args[index]).strip()
|
||
lowered = token.lower()
|
||
option_name = lowered.split("=", 1)[0]
|
||
if option_name in profile_options:
|
||
raise GrokBuildError(
|
||
f"{option_name} 可加载带独立模型的 Agent 配置，"
|
||
"禁止在受管密钥模式下使用"
|
||
)
|
||
if option_name in model_options:
|
||
if "=" in token:
|
||
requested = token.split("=", 1)[1].strip()
|
||
index += 1
|
||
else:
|
||
if index + 1 >= len(args):
|
||
raise GrokBuildError(f"{token} 缺少模型参数")
|
||
requested = str(args[index + 1]).strip()
|
||
index += 2
|
||
if requested != MODEL_PROFILE:
|
||
raise GrokBuildError(
|
||
f"{option_name} 只允许后台受管模型 {MODEL_PROFILE}"
|
||
)
|
||
continue
|
||
index += 1
|
||
|
||
|
||
def _build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(
|
||
description="当前项目的 Grok Build 运行时、模型同步和启动入口"
|
||
)
|
||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||
subparsers.add_parser("status", help="输出运行时和模型同步状态")
|
||
|
||
install = subparsers.add_parser("install", help="安装官方 Grok Build 运行时")
|
||
install.add_argument("--version", default="", help="指定版本；留空安装 stable")
|
||
|
||
sync = subparsers.add_parser(
|
||
"sync",
|
||
help="从 ai_settings.local.json 同步模型",
|
||
)
|
||
sync.add_argument(
|
||
"--include-mcp",
|
||
action="store_true",
|
||
help="同时导入已启用的客服 MCP；默认关闭以防止扩大数据权限",
|
||
)
|
||
|
||
tui = subparsers.add_parser("tui", help="打开完整原生 TUI")
|
||
tui.add_argument("--cwd", default=str(PROJECT_DIR))
|
||
tui.add_argument("--model", default="")
|
||
tui.add_argument("prompt", nargs="?", default="")
|
||
|
||
login = subparsers.add_parser(
|
||
"login",
|
||
help="已禁用：本项目只使用后台自有模型，不建立 xAI 登录",
|
||
)
|
||
login.add_argument("--cwd", default=str(PROJECT_DIR))
|
||
|
||
run = subparsers.add_parser("run", help="执行无头任务")
|
||
run.add_argument("prompt")
|
||
run.add_argument("--cwd", default=str(PROJECT_DIR))
|
||
run.add_argument("--model", default="")
|
||
run.add_argument("--effort", default="high")
|
||
run.add_argument("--max-turns", type=int, default=50)
|
||
run.add_argument("--yolo", action="store_true")
|
||
run.add_argument("--read-only", action="store_true")
|
||
run.add_argument("--continue", dest="continue_session", action="store_true")
|
||
run.add_argument("--resume", default="")
|
||
run.add_argument("--sandbox", default="")
|
||
run.add_argument("--tools", default="")
|
||
run.add_argument("--disallowed-tools", default="")
|
||
run.add_argument("--rules", default="")
|
||
|
||
acp = subparsers.add_parser("acp", help="启动 ACP JSON-RPC stdio 服务")
|
||
acp.add_argument("--cwd", default=str(PROJECT_DIR))
|
||
acp.add_argument("--model", default="")
|
||
acp.add_argument("--yolo", action="store_true")
|
||
acp.add_argument("--sandbox", default="")
|
||
|
||
passthrough = subparsers.add_parser("exec", help="将剩余参数原样传给 Grok")
|
||
passthrough.add_argument("--cwd", default=str(PROJECT_DIR))
|
||
passthrough.add_argument(
|
||
"--with-managed-secrets",
|
||
action="store_true",
|
||
help="未知上游命令需要后台模型/MCP 密钥时显式启用；wrap 永不允许",
|
||
)
|
||
passthrough.add_argument(
|
||
"--allow-unknown",
|
||
action="store_true",
|
||
help="允许未知上游管理命令运行，但不注入后台模型/MCP 密钥",
|
||
)
|
||
passthrough.add_argument("args", nargs=argparse.REMAINDER)
|
||
return parser
|
||
|
||
|
||
def main(argv: Iterable[str] | None = None) -> int:
|
||
args = _build_parser().parse_args(list(argv) if argv is not None else None)
|
||
manager = GrokBuildManager()
|
||
try:
|
||
if args.command == "status":
|
||
print(json.dumps(_status_dict(manager.status()), ensure_ascii=False, indent=2))
|
||
return 0
|
||
if args.command == "install":
|
||
def progress(received: int, total: int) -> None:
|
||
if total:
|
||
percent = min(100, int(received * 100 / total))
|
||
print(f"\r下载中 {percent:3d}% {received}/{total} 字节", end="", flush=True)
|
||
else:
|
||
print(f"\r下载中 {received} 字节", end="", flush=True)
|
||
|
||
status = manager.install_official_release(args.version or None, progress)
|
||
print()
|
||
print(json.dumps(_status_dict(status), ensure_ascii=False, indent=2))
|
||
return 0
|
||
if args.command == "sync":
|
||
result = manager.sync_model_configuration(include_mcp=args.include_mcp)
|
||
print(json.dumps(asdict(result), ensure_ascii=False, indent=2))
|
||
return 0 if result.compatible else 2
|
||
if args.command == "tui":
|
||
process = manager.open_tui(
|
||
workspace=args.cwd,
|
||
initial_prompt=args.prompt,
|
||
model=args.model,
|
||
)
|
||
if manager.model_profile().api_backend == "dify":
|
||
try:
|
||
return int(process.wait())
|
||
except KeyboardInterrupt:
|
||
try:
|
||
process.terminate()
|
||
return int(process.wait(timeout=3))
|
||
except (OSError, subprocess.SubprocessError):
|
||
try:
|
||
process.kill()
|
||
except OSError:
|
||
pass
|
||
return 130
|
||
return 0
|
||
if args.command == "login":
|
||
raise GrokBuildError(
|
||
"本项目已禁用 Grok/xAI 登录；Grok Build 仅作为 Agent，"
|
||
"请在后台配置自有模型"
|
||
)
|
||
if args.command == "run":
|
||
binary = manager.require_binary()
|
||
manager.prepare_agent_configuration()
|
||
requested_model = args.model.strip()
|
||
if requested_model and requested_model != MODEL_PROFILE:
|
||
raise GrokBuildError(
|
||
f"本项目的 Grok Agent 只允许使用后台受管模型 {MODEL_PROFILE}"
|
||
)
|
||
if args.read_only:
|
||
manager.verify_read_only_environment(args.cwd)
|
||
command = manager.build_headless_args(
|
||
args.prompt,
|
||
workspace=args.cwd,
|
||
model=MODEL_PROFILE,
|
||
effort=args.effort,
|
||
max_turns=args.max_turns,
|
||
auto_approve=args.yolo,
|
||
read_only=args.read_only,
|
||
continue_session=args.continue_session,
|
||
resume_session=args.resume,
|
||
sandbox=args.sandbox,
|
||
allowed_tools=args.tools,
|
||
disallowed_tools=args.disallowed_tools,
|
||
rules=args.rules,
|
||
)
|
||
completed = subprocess.run(
|
||
[str(binary), *command],
|
||
cwd=str(Path(args.cwd).resolve()),
|
||
env=manager.runtime_environment(
|
||
include_model_key=True,
|
||
include_mcp_secrets=not args.read_only,
|
||
workspace=args.cwd,
|
||
custom_model_only=True,
|
||
),
|
||
check=False,
|
||
)
|
||
return int(completed.returncode)
|
||
if args.command == "acp":
|
||
binary = manager.require_binary()
|
||
manager.prepare_agent_configuration()
|
||
requested_model = args.model.strip()
|
||
if requested_model and requested_model != MODEL_PROFILE:
|
||
raise GrokBuildError(
|
||
f"本项目的 Grok Agent 只允许使用后台受管模型 {MODEL_PROFILE}"
|
||
)
|
||
command = manager.build_acp_args(
|
||
model=MODEL_PROFILE,
|
||
yolo=args.yolo,
|
||
sandbox=args.sandbox,
|
||
)
|
||
completed = subprocess.run(
|
||
[str(binary), *command],
|
||
cwd=str(Path(args.cwd).resolve()),
|
||
env=manager.runtime_environment(
|
||
include_model_key=True,
|
||
include_mcp_secrets=True,
|
||
workspace=args.cwd,
|
||
custom_model_only=True,
|
||
),
|
||
check=False,
|
||
)
|
||
return int(completed.returncode)
|
||
if args.command == "exec":
|
||
binary = manager.require_binary()
|
||
forwarded = normalize_passthrough_args(args.args)
|
||
process_workspace = Path(args.cwd).resolve()
|
||
effective_workspace = resolve_passthrough_workspace(
|
||
forwarded,
|
||
process_workspace,
|
||
)
|
||
passthrough_kind = classify_passthrough_args(forwarded)
|
||
if passthrough_kind == "safe" and any(
|
||
str(token).strip().lower() in {"login", "logout", "setup"}
|
||
for token in forwarded
|
||
):
|
||
raise GrokBuildError(
|
||
"本项目禁止通过 Grok 建立或修改 xAI 登录；"
|
||
"请使用后台自有模型配置"
|
||
)
|
||
if (
|
||
passthrough_kind in {"wrap", "extension"}
|
||
and args.with_managed_secrets
|
||
):
|
||
surface = (
|
||
"wrap"
|
||
if passthrough_kind == "wrap"
|
||
else "--plugin-dir/--agent-profile"
|
||
)
|
||
raise GrokBuildError(
|
||
f"{surface} 会加载任意本地代码，禁止向其注入受管密钥"
|
||
)
|
||
if (
|
||
passthrough_kind == "unknown"
|
||
and not args.with_managed_secrets
|
||
and not args.allow_unknown
|
||
):
|
||
raise GrokBuildError(
|
||
"未知 Grok 命令默认不执行，以免凭据回退或泄漏；"
|
||
"管理命令使用 exec --allow-unknown -- <参数>，"
|
||
"确认需要模型能力后使用 exec --with-managed-secrets -- <参数>"
|
||
)
|
||
include_secrets = (
|
||
passthrough_kind == "agent" or args.with_managed_secrets
|
||
) and passthrough_kind not in {"wrap", "extension"}
|
||
if include_secrets:
|
||
manager.prepare_agent_configuration()
|
||
reject_passthrough_model_bypasses(forwarded)
|
||
if passthrough_kind == "agent":
|
||
forwarded = enforce_passthrough_agent_model(forwarded)
|
||
if "--no-auto-update" not in forwarded:
|
||
forwarded.insert(0, "--no-auto-update")
|
||
completed = subprocess.run(
|
||
[str(binary), *forwarded],
|
||
cwd=str(process_workspace),
|
||
env=manager.runtime_environment(
|
||
include_model_key=include_secrets,
|
||
include_mcp_secrets=include_secrets,
|
||
workspace=effective_workspace,
|
||
custom_model_only=True,
|
||
),
|
||
check=False,
|
||
)
|
||
return int(completed.returncode)
|
||
except (GrokBuildError, OSError, subprocess.SubprocessError) as exc:
|
||
print(f"错误：{exc}", file=sys.stderr)
|
||
return 1
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|