# -*- 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 ''}" 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())