projectAIpopular/router_system/tools.py

"""工具调用内核 —— 让 LLM 以 OpenAI function-calling 协议操作工作区文件。

组成（对齐《实现方案_v4_模型池与工具智能体.md》D4）：
- TOOLS_SPEC：list_dir / read_file / write_file / edit_file / search_files /
  run_command / web_fetch 七个工具的 OpenAI tools 声明
- WorkspaceTools：被"关押"在根目录内的文件工具（路径越界一律拒绝，Windows pathlib）；
  web_fetch 带公网 SSRF 防护（dsh web_fetch 同款），run_command 带危险命令拦截
- parse_tool_calls：解析 OpenAI 响应里的 tool_calls（arguments 容错为 {}）
- ToolLoop：通用智能体循环。chat_fn 注入（网关传 OpenAI 兼容客户端，测试传假实现），
  本模块只负责循环编排：调用 -> 执行工具 -> 回喂结果 -> 直到模型给出最终答复。

工程约束：
- 纯标准库（router_system 零第三方依赖不变）
- 工具结果回喂前截断（防止上下文爆炸），轮数与 token 双上限（金额护栏）
- 事件回调 on_event 逐条产出过程事件（供 SSE 透出"智能体在做什么"）
- 工具在线程池执行（asyncio.to_thread），长命令/大搜索不阻塞事件循环
- 重复同参调用达到阈值回喂警语（防模型原地打转，dsh repeat-reminder 同款）
"""
from __future__ import annotations

import asyncio
import ipaddress
import json
from pathlib import Path
from typing import Any, Awaitable, Callable, Dict, List, Optional

# 回喂给模型的工具结果/读取内容上限（字符）
MAX_READ_CHARS = 8000
MAX_LIST_ENTRIES = 200
MAX_RESULT_CHARS = 8000
MAX_WRITE_CHARS = 200_000

# search_files 上限
SEARCH_MAX_MATCHES = 30
SEARCH_MAX_FILES = 400
SEARCH_MAX_FILE_BYTES = 512 * 1024
SEARCH_SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", "dist", "build"}

# run_command 上限
SHELL_OUTPUT_CHARS = 4000

# web_fetch 上限（SSRF 防护：仅公网 http/https，拒绝私网/环回/链路本地地址）
FETCH_MAX_CHARS = 12000
FETCH_TIMEOUT_S = 15.0
FETCH_MAX_BYTES = 512 * 1024
FETCH_ALLOWED_SCHEMES = ("http", "https")
# NAT64 Well-Known Prefix（dsh 同款拒绝项）
_NAT64_PREFIX = ipaddress.ip_network("64:ff9b::/96")

# 重复调用提醒阈值（同一工具 + 同一参数第 N 次起提示模型换策略）
REPEAT_CALL_WARN_AT = 3

# 默认循环上限
DEFAULT_MAX_ROUNDS = 8

# 可能长时间运行的工具（命令执行 / 网络抓取 / 大范围搜索）放独立线程执行，
# 不阻塞事件循环；完成等待用 asyncio.sleep 轮询（补丁运行时的 TestClient
# 每请求独立事件循环，不推进 run_in_executor 桥接；轮询在生产/测试两端都可靠）。
SLOW_TOOLS = {"run_command", "web_fetch", "search_files"}


def _spawn_tool_thread(fn, *args):
    """起守护线程执行 fn(*args)，返回 (结果盒子, 线程)；轮询线程存活后取盒内值。"""
    import threading
    box: Dict[str, Any] = {}

    def _runner():
        try:
            box["result"] = fn(*args)
        except BaseException as exc:   # 线程内异常回传给调用方
            box["error"] = exc

    t = threading.Thread(target=_runner, daemon=True, name="agenttool")
    t.start()
    return box, t


def _dispatch_tool(tools, name, arguments):
    """统一工具分发（内部辅助）。"""
    return tools.execute(name, arguments)


async def run_tool_async(tools: "WorkspaceTools", name: str,
                         arguments: Dict[str, Any]) -> Dict[str, Any]:
    """执行一个工具：慢工具放线程并轮询等完成，快工具直接内联执行。"""
    if name not in SLOW_TOOLS:
        return _dispatch_tool(tools, name, arguments)
    box, t = _spawn_tool_thread(_dispatch_tool, tools, name, arguments)
    while t.is_alive():
        await asyncio.sleep(0.02)
    if "error" in box:
        raise box["error"]
    return box["result"]

# OpenAI tools 声明（chat/completions 请求的 tools 参数）
TOOLS_SPEC: List[Dict[str, Any]] = [
    {
        "type": "function",
        "function": {
            "name": "list_dir",
            "description": "列出工作区内目录的内容（文件与子目录，含大小）。",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "工作区内的相对路径，默认根目录"}
                },
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取工作区内一个文本文件的内容（过长自动截断）。",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "工作区内的相对路径"}
                },
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "把文本内容写入（或创建/覆盖）工作区内的一个文件。",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "工作区内的相对路径"},
                    "content": {"type": "string", "description": "要写入的全文"},
                },
                "required": ["path", "content"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "edit_file",
            "description": "对工作区内已有文件做精确替换编辑：old_string 必须在文件中恰好出现一次，"
                           "被替换为 new_string。适合小改动；大改用 write_file 整体重写。",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "工作区内的相对路径"},
                    "old_string": {"type": "string", "description": "要替换的原文（须唯一匹配）"},
                    "new_string": {"type": "string", "description": "替换后的新文"},
                },
                "required": ["path", "old_string", "new_string"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_files",
            "description": "在工作区（或其子目录）内按关键词跨文件搜索文本内容，"
                           "返回匹配的文件/行号/行内容（自动跳过 .git、node_modules 等目录与二进制大文件）。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索的关键词（大小写敏感）"},
                    "path": {"type": "string", "description": "限定的子目录，默认整个工作区"},
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "run_command",
            "description": "在工作区根目录执行一条 shell 命令并返回退出码与输出（如运行测试、查看版本）。"
                           "仅当系统开启 allow_shell 时可用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {"type": "string", "description": "要执行的命令行"},
                },
                "required": ["command"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "web_fetch",
            "description": "抓取一个公网 http/https URL 的文本内容（如查文档/接口说明），"
                           "返回截断后的正文。私网/环回地址会被拒绝；仅当系统开启 allow_net 时可用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "要抓取的完整 URL（http/https）"},
                },
                "required": ["url"],
            },
        },
    },
]

TOOL_NAMES = {t["function"]["name"] for t in TOOLS_SPEC}


def browse_directories(path: str = "") -> Dict[str, Any]:
    """目录选择器的本地文件系统浏览（只列目录，不读文件内容）。

    path 为空时列出 Windows 盘符（POSIX 列根目录）。返回
    {"ok", "path", "parent", "dirs": [名称]}；用于智能体"选择工作区"。
    """
    import os
    if not path or not path.strip():
        if os.name == "nt":
            drives = [f"{c}:\\" for c in "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
                      if os.path.exists(f"{c}:\\")]
            return {"ok": True, "path": "", "parent": "", "dirs": drives}
        return {"ok": True, "path": "/", "parent": "",
                "dirs": sorted(os.listdir("/"))}
    p = Path(path).resolve()
    if not p.exists():
        return {"ok": False, "error": f"路径不存在: {path}"}
    if not p.is_dir():
        return {"ok": False, "error": f"不是目录: {path}"}
    dirs = []
    for child in sorted(p.iterdir(), key=lambda c: c.name.lower()):
        try:
            if child.is_dir():
                dirs.append(child.name)
        except OSError:
            continue  # 无权限/符号链接坏点，跳过
    parent = str(p.parent) if p.parent != p else ""
    return {"ok": True, "path": str(p), "parent": parent, "dirs": dirs}


class ToolError(Exception):
    """工具执行失败（路径越界/不存在/参数非法）。"""


def _atomic_write_text(p: Path, text: str) -> None:
    """原子写文本：同目录临时文件 + os.replace（防半截文件；dsh atomic-write 同款）。

    Windows 上目标被占用时 os.replace 可能 EPERM：小退避重试一次，
    仍失败则退回直接写（保可用性，牺牲原子性）。
    """
    import os
    import tempfile
    import time
    tmp = None
    try:
        with tempfile.NamedTemporaryFile(
                "w", encoding="utf-8", dir=str(p.parent),
                prefix=p.name + ".", suffix=".tmp", delete=False) as f:
            tmp = Path(f.name)
            f.write(text)
        for attempt in (0, 1):
            try:
                os.replace(tmp, p)
                return
            except PermissionError:
                if attempt == 0:
                    time.sleep(0.05)
        raise
    except PermissionError:
        if tmp is not None:
            tmp.unlink(missing_ok=True)
        p.write_text(text, encoding="utf-8")  # 退路：非原子但保可用
    except BaseException:
        if tmp is not None:
            tmp.unlink(missing_ok=True)
        raise


# 危险命令模式（大小写不敏感）：宁可误拦不可漏拦（用户可换写法绕开误拦项）
_DANGEROUS_PATTERNS = [
    (r"\bformat\b\s+[a-z]:", "格式化磁盘"),
    (r"\brd\s+/s", "递归删除目录"),
    (r"\brmdir\s+/s", "递归删除目录"),
    (r"\bdel\s+/[fsmq]", "强制/递归删除"),
    (r"\brm\s+(-[a-z]*r[a-z]*f|-[a-z]*f[a-z]*r)\s+[/~]", "递归强制删除根/家目录"),
    (r"\bshutdown\b", "关机/重启"),
    (r"\bdiskpart\b", "磁盘分区操作"),
    (r"\bbcdedit\b", "启动配置修改"),
    (r"\breg\s+delete\b", "注册表删除"),
    (r"\bvssadmin\s+delete\b", "卷影副本删除"),
    (r"\bmkfs\b", "格式化文件系统"),
    (r"\bdd\s+if=", "裸磁盘写入"),
    (r":\(\)\s*\{.*\};\s*:", "fork 炸弹"),
    (r"\bcurl\b[^|]*\|\s*(ba)?sh\b", "下载并执行脚本"),
    (r"\bwget\b[^|]*\|\s*(ba)?sh\b", "下载并执行脚本"),
    (r"\biwr\b[^|]*\|\s*iex\b", "下载并执行脚本"),
]


def _dangerous_command_reason(command: str) -> str:
    """命中危险命令模式时返回原因，否则返回空串（审批之外的独立防线）。"""
    import re
    lowered = command.lower()
    for pattern, reason in _DANGEROUS_PATTERNS:
        if re.search(pattern, lowered):
            return reason
    return ""


class WorkspaceTools:
    """被限制在根目录内的文件工具（智能体的"手"）。

    安全：所有路径先 join 再 resolve，解析结果必须仍位于根目录内
    （根目录自身允许），否则抛 ToolError——防 ../ 越界与绝对路径逃逸。
    """

    def __init__(self, root: str | Path,
                 allow_shell: bool = False, shell_timeout_s: int = 20,
                 allow_net: bool = True):
        self.root = Path(root).resolve()
        self.root.mkdir(parents=True, exist_ok=True)
        self.allow_shell = bool(allow_shell)
        self.shell_timeout_s = max(1, int(shell_timeout_s))
        self.allow_net = bool(allow_net)

    # ---------- 路径关押 ----------
    def resolve(self, rel_path: str) -> Path:
        rel = (rel_path or "").strip().replace("\\", "/").lstrip("/")
        p = (self.root / rel).resolve()
        if p != self.root and self.root not in p.parents:
            raise ToolError(f"路径越界（不允许访问工作区之外）: {rel_path}")
        return p

    # ---------- 三个工具 ----------
    def list_dir(self, rel_path: str = "") -> Dict[str, Any]:
        d = self.resolve(rel_path)
        if not d.exists():
            return {"ok": False, "error": f"目录不存在: {rel_path}"}
        if not d.is_dir():
            return {"ok": False, "error": f"不是目录: {rel_path}"}
        entries = []
        for child in sorted(d.iterdir(), key=lambda c: (c.is_file(), c.name.lower())):
            if child.is_dir():
                entries.append({"name": child.name + "/", "type": "dir"})
            else:
                entries.append({
                    "name": child.name, "type": "file",
                    "size": child.stat().st_size,
                })
            if len(entries) >= MAX_LIST_ENTRIES:
                entries.append({"name": f"…（超过 {MAX_LIST_ENTRIES} 项已截断）", "type": "notice"})
                break
        return {"ok": True, "path": rel_path or ".", "entries": entries}

    def read_file(self, rel_path: str) -> Dict[str, Any]:
        p = self.resolve(rel_path)
        if not p.exists():
            return {"ok": False, "error": f"文件不存在: {rel_path}"}
        if not p.is_file():
            return {"ok": False, "error": f"不是文件: {rel_path}"}
        try:
            text = p.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            return {"ok": False, "error": f"非文本文件（UTF-8 解码失败）: {rel_path}"}
        truncated = len(text) > MAX_READ_CHARS
        return {
            "ok": True,
            "path": rel_path,
            "content": text[:MAX_READ_CHARS],
            "truncated": truncated,
            "total_chars": len(text),
        }

    def write_file(self, rel_path: str, content: str) -> Dict[str, Any]:
        if len(content) > MAX_WRITE_CHARS:
            return {"ok": False, "error": f"内容过长（>{MAX_WRITE_CHARS} 字符），拒绝写入"}
        p = self.resolve(rel_path)
        p.parent.mkdir(parents=True, exist_ok=True)
        _atomic_write_text(p, content)
        return {"ok": True, "path": rel_path, "bytes_written": len(content.encode("utf-8"))}

    def edit_file(self, rel_path: str, old_string: str, new_string: str) -> Dict[str, Any]:
        """精确替换编辑：old_string 必须在文件中恰好出现一次（harness 式安全编辑）。"""
        if not old_string:
            return {"ok": False, "error": "old_string 不能为空"}
        if len(old_string) > MAX_READ_CHARS:
            return {"ok": False, "error": "old_string 过长（先 read_file 分段定位）"}
        p = self.resolve(rel_path)
        if not p.exists() or not p.is_file():
            return {"ok": False, "error": f"文件不存在: {rel_path}"}
        try:
            text = p.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            return {"ok": False, "error": f"非文本文件: {rel_path}"}
        count = text.count(old_string)
        if count == 0:
            return {"ok": False, "error": "old_string 未在文件中找到（先 read_file 核对原文）"}
        if count > 1:
            return {"ok": False,
                    "error": f"old_string 出现 {count} 次（要求唯一）；请扩大上下文使其唯一"}
        new_text = text.replace(old_string, new_string, 1)
        _atomic_write_text(p, new_text)
        return {
            "ok": True, "path": rel_path,
            "replaced": 1,
            "changed_chars": len(new_text) - len(text),
        }

    def search_files(self, query: str, rel_path: str = "") -> Dict[str, Any]:
        """跨文件文本搜索（os.walk 修剪依赖/构建目录，限量返回，不跟随符号链接）。"""
        import os
        if not query:
            return {"ok": False, "error": "query 不能为空"}
        base = self.resolve(rel_path or "")
        if not base.exists() or not base.is_dir():
            return {"ok": False, "error": f"目录不存在: {rel_path}"}
        matches: List[Dict[str, Any]] = []
        scanned = 0
        truncated = False

        def _match_file(p: Path) -> bool:
            """在单文件内找匹配（找到即 True）。"""
            nonlocal matches, truncated
            for lineno, line in enumerate(p.read_text(encoding="utf-8").splitlines(), 1):
                if query in line:
                    rel = p.relative_to(self.root).as_posix()
                    matches.append({
                        "file": rel, "line": lineno,
                        "text": line.strip()[:300],
                    })
                    if len(matches) >= SEARCH_MAX_MATCHES:
                        truncated = True
                        return True
            return False

        for dirpath, dirnames, filenames in os.walk(base, followlinks=False):
            # 修剪依赖/构建目录：不进入（rglob 全量物化在大工作区上不可接受）
            dirnames[:] = sorted((d for d in dirnames if d not in SEARCH_SKIP_DIRS),
                                 key=str.lower)
            if len(matches) >= SEARCH_MAX_MATCHES or scanned > SEARCH_MAX_FILES:
                truncated = True
                break
            for fname in sorted(filenames, key=str.lower):
                fpath = Path(dirpath) / fname
                try:
                    if not fpath.is_file():
                        continue
                    if fpath.stat().st_size > SEARCH_MAX_FILE_BYTES:
                        continue
                    scanned += 1
                    if scanned > SEARCH_MAX_FILES:
                        truncated = True
                        break
                    if _match_file(fpath):
                        break
                except (OSError, UnicodeDecodeError):
                    continue  # 二进制/不可读/并发删除，跳过
            if len(matches) >= SEARCH_MAX_MATCHES:
                truncated = True
                break
        return {"ok": True, "query": query, "matches": matches,
                "scanned_files": scanned, "truncated": truncated}

    def run_command(self, command: str) -> Dict[str, Any]:
        """在工作区根目录执行一条 shell 命令行（默认关闭，allow_shell 开启后可用）。

        安全设计（dsh bash 工具同款语义）：
        - 危险命令模式先拦截（即使审批通过也拒绝）
        - 用显式 shell 解释器的参数列表执行（cmd /c 或 /bin/sh -c），
          命令字符串对解释器可见属于功能本体，防护依赖 allow_shell 开关
          + 审批门卫 + 超时熔断 + cwd 关押在本工作区
        """
        import os
        import subprocess
        if not self.allow_shell:
            return {"ok": False,
                    "error": "run_command 未启用（系统设置 allow_shell 为关）。"
                             "请让用户在智能体页打开「允许执行命令」后重试。"}
        command = (command or "").strip()
        if not command:
            return {"ok": False, "error": "command 不能为空"}
        blocked = _dangerous_command_reason(command)
        if blocked:
            return {"ok": False, "error": f"命令被安全策略拒绝（{blocked}）。请换一种不具破坏性的做法。"}
        if os.name == "nt":
            # Windows：显式走 cmd /c（与 shell=True 内部同构——整条命令包一层引号，
            # 避免参数列表的 CRT 转义与 cmd 引号语义冲突）；COMSPEC 取系统 shell 路径
            comspec = os.environ.get("COMSPEC", "cmd.exe")
            run_args: Any = f'"{comspec}" /c "{command}"'
            creationflags = 0x08000000  # CREATE_NO_WINDOW
        else:
            run_args = ["/bin/sh", "-c", command]
            creationflags = 0
        try:
            proc = subprocess.run(
                run_args, cwd=str(self.root), capture_output=True,
                timeout=self.shell_timeout_s, creationflags=creationflags,
            )
            out = (proc.stdout or b"").decode("utf-8", errors="replace")
            err = (proc.stderr or b"").decode("utf-8", errors="replace")
            combined = (out + ("\n[stderr]\n" + err if err.strip() else "")).strip()
            if len(combined) > SHELL_OUTPUT_CHARS:
                combined = combined[:SHELL_OUTPUT_CHARS] + "…(输出截断)"
            return {"ok": True, "exit_code": proc.returncode,
                    "output": combined or "(无输出)", "command": command}
        except subprocess.TimeoutExpired:
            return {"ok": False, "error": f"命令超时（>{self.shell_timeout_s}s），已终止: {command}"}
        except OSError as e:
            return {"ok": False, "error": f"命令执行失败: {type(e).__name__}: {e}"}

    def web_fetch(self, url: str) -> Dict[str, Any]:
        """抓取公网 URL 文本（SSRF 防护：拒绝非 http(s)、私网/环回/链路本地/NAT64 目标）。

        校验流程对齐 dsh web_fetch：解析 DNS -> 全部地址必须公网 -> 才发起请求；
        响应限量（字节/字符）、超时熔断、二进制嗅探拒绝。
        """
        import socket
        import urllib.error
        import urllib.parse
        import urllib.request

        if not self.allow_net:
            return {"ok": False, "error": "web_fetch 未启用（系统设置 allow_net 为关）。"}
        raw = (url or "").strip()
        try:
            parsed = urllib.parse.urlsplit(raw)
        except ValueError:
            return {"ok": False, "error": f"URL 无法解析: {raw[:120]}"}
        if parsed.scheme.lower() not in FETCH_ALLOWED_SCHEMES:
            return {"ok": False, "error": f"仅允许 http/https URL（收到 {parsed.scheme or '空'}）"}
        host = parsed.hostname or ""
        if not host:
            return {"ok": False, "error": "URL 缺少主机名"}
        try:
            port = parsed.port
        except ValueError:
            return {"ok": False, "error": "URL 端口非法"}
        # DNS 解析后逐一校验：任何私网/环回/链路本地/保留/NAT64 地址都拒绝
        try:
            infos = socket.getaddrinfo(host, port or (443 if parsed.scheme == "https" else 80),
                                       proto=socket.IPPROTO_TCP)
        except socket.gaierror as e:
            return {"ok": False, "error": f"域名解析失败: {host} ({e})"}
        for info in infos:
            ip = info[4][0]
            try:
                addr = ipaddress.ip_address(ip.split("%")[0])  # 剥 zone id
            except ValueError:
                return {"ok": False, "error": f"解析出非法地址: {ip}"}
            if (addr.is_private or addr.is_loopback or addr.is_link_local
                    or addr.is_reserved or addr.is_multicast or addr.is_unspecified
                    or (addr.version == 6 and addr in _NAT64_PREFIX)):
                return {"ok": False,
                        "error": f"目标地址 {addr} 属于内网/保留段，已被 SSRF 防护拒绝"}
        try:
            req = urllib.request.Request(raw, headers={"User-Agent": "router-agent/1.0"})
            with urllib.request.urlopen(req, timeout=FETCH_TIMEOUT_S) as resp:
                body = resp.read(FETCH_MAX_BYTES + 1)
                charset = resp.headers.get_content_charset() or "utf-8"
        except urllib.error.HTTPError as e:
            return {"ok": False, "error": f"HTTP {e.code}: {e.reason}"}
        except (urllib.error.URLError, OSError, ValueError) as e:
            return {"ok": False, "error": f"抓取失败: {type(e).__name__}: {e}"}
        if len(body) > FETCH_MAX_BYTES:
            return {"ok": False, "error": f"响应超过 {FETCH_MAX_BYTES // 1024}KB 上限，拒绝处理"}
        if b"\x00" in body[:512]:
            return {"ok": False, "error": "非文本内容（检测到二进制），拒绝处理"}
        try:
            text = body.decode(charset, errors="replace")
        except LookupError:
            text = body.decode("utf-8", errors="replace")
        truncated = len(text) > FETCH_MAX_CHARS
        return {"ok": True, "url": raw,
                "content": text[:FETCH_MAX_CHARS], "truncated": truncated,
                "total_chars": len(text)}

    # ---------- 统一执行入口 ----------
    def execute(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """按名字执行工具；任何异常折叠为 {"ok": False, "error": ...}。"""
        try:
            if name == "list_dir":
                return self.list_dir(str(arguments.get("path", "")))
            if name == "read_file":
                return self.read_file(str(arguments.get("path", "")))
            if name == "write_file":
                return self.write_file(
                    str(arguments.get("path", "")), str(arguments.get("content", "")))
            if name == "edit_file":
                return self.edit_file(
                    str(arguments.get("path", "")),
                    str(arguments.get("old_string", "")),
                    str(arguments.get("new_string", "")))
            if name == "search_files":
                return self.search_files(
                    str(arguments.get("query", "")), str(arguments.get("path", "")))
            if name == "run_command":
                return self.run_command(str(arguments.get("command", "")))
            if name == "web_fetch":
                return self.web_fetch(str(arguments.get("url", "")))
            return {"ok": False, "error": f"未知工具: {name}"}
        except ToolError as e:
            return {"ok": False, "error": str(e)}
        except OSError as e:
            return {"ok": False, "error": f"文件系统错误: {type(e).__name__}: {e}"}


def _loads_json_object(text: str) -> Dict[str, Any]:
    """宽松解析 JSON 对象：剥除 markdown 围栏、取首个 {...}；失败返回 {}。"""
    t = (text or "").strip()
    fence = "`" * 3
    t = t.replace(fence + "json", fence).replace(fence, "").strip()
    try:
        obj = json.loads(t)
        if isinstance(obj, dict):
            return obj
    except json.JSONDecodeError:
        pass
    start, end = t.find("{"), t.rfind("}")
    if start != -1 and end > start:
        try:
            obj = json.loads(t[start:end + 1])
            if isinstance(obj, dict):
                return obj
        except json.JSONDecodeError:
            pass
    return {}


def parse_tool_calls(message: Dict[str, Any]) -> List[Dict[str, Any]]:
    """从 OpenAI 响应的 message 解析 tool_calls。

    返回 [{"id","name","arguments"(dict)}]；arguments 非法 JSON 时容错为 {}。
    """
    out: List[Dict[str, Any]] = []
    for tc in message.get("tool_calls") or []:
        fn = tc.get("function") or {}
        raw = fn.get("arguments")
        if isinstance(raw, dict):
            args = raw
        elif isinstance(raw, str) and raw.strip():
            try:
                parsed = json.loads(raw)
                args = parsed if isinstance(parsed, dict) else {}
            except json.JSONDecodeError:
                args = {}
        else:
            args = {}
        out.append({
            "id": tc.get("id") or f"call_{len(out)}",
            "name": fn.get("name") or "",
            "arguments": args,
        })
    return out


class ToolLoop:
    """通用智能体工具循环（zcode 式：思考 -> 调工具 -> 看结果 -> 再思考）。

    chat_fn(messages, tools_spec) -> {"content": str|None,
                                      "tool_calls": [ {id,name,arguments}, ... ],
                                      "usage": {"prompt_tokens", "completion_tokens"}}
      由网关注入真实 OpenAI 兼容客户端；测试注入脚本化假实现。

    on_event(ev) 为可选同步回调，逐条收到过程事件：
      {"type":"round","round":n}
      {"type":"tool_call","round":n,"name":...,"arguments":...}
      {"type":"tool_result","round":n,"name":...,"ok":...,"preview":...}
      {"type":"usage","prompt_tokens":...,"completion_tokens":...}
      {"type":"final","round":n,"reason":"answer"|"max_rounds"|"token_cap"|"error"}
    """

    def __init__(
        self,
        tools: WorkspaceTools,
        chat_fn: Callable[[List[Dict[str, Any]], List[Dict[str, Any]]], Awaitable[Dict[str, Any]]],
        max_rounds: int = DEFAULT_MAX_ROUNDS,
        token_cap: int = 0,
        on_event: Optional[Callable[[Dict[str, Any]], None]] = None,
        result_preview_chars: int = MAX_RESULT_CHARS,
        emit_final: bool = True,
        approval_hook: Optional[Callable[[str, Dict[str, Any]], Awaitable[bool]]] = None,
        on_delta: Optional[Callable[[str], None]] = None,
    ):
        self.tools = tools
        self.chat_fn = chat_fn
        self.max_rounds = max(1, int(max_rounds))
        self.token_cap = int(token_cap)          # 0 = 不限
        self.on_event = on_event
        self.result_preview_chars = result_preview_chars
        self.emit_final = emit_final             # 两级模式内层循环置 False，由外层统一收尾
        # 审批门卫（D9）：执行工具前调用，返回 False = 用户拒绝（可选；缺省跳过审批）
        self.approval_hook = approval_hook
        # 流式增量回调（D10，可选）：chat_fn 支持 on_delta 参数时逐段转发模型文本
        self.on_delta = on_delta
        self._accepts_delta: Optional[bool] = None
        # 重复调用计数（dsh repeat-tool-reminder 同款提醒，防模型原地打转）
        self._call_counts: Dict[tuple, int] = {}

    def _emit(self, ev: Dict[str, Any]) -> None:
        if ev.get("type") == "final" and not self.emit_final:
            return  # 内层循环不发终态事件（外层编排负责）
        if self.on_event is not None:
            try:
                self.on_event(ev)
            except Exception:
                pass  # 事件回调不允许打断主循环

    def _total_tokens(self, usage: Dict[str, int]) -> int:
        return int(usage.get("prompt_tokens", 0)) + int(usage.get("completion_tokens", 0))

    def _invoke_chat(self, messages: List[Dict[str, Any]]) -> Awaitable[Dict[str, Any]]:
        """调用 chat_fn；其签名支持 on_delta 时才传入（对旧假实现向后兼容）。"""
        if self._accepts_delta is None:
            try:
                import inspect
                self._accepts_delta = len(inspect.signature(self.chat_fn).parameters) >= 3
            except (TypeError, ValueError):
                self._accepts_delta = False
        if self.on_delta is not None and self._accepts_delta:
            return self.chat_fn(messages, TOOLS_SPEC, self.on_delta)
        return self.chat_fn(messages, TOOLS_SPEC)

    async def run(self, task: str, system: str = "",
                  history: Optional[List[Dict[str, Any]]] = None) -> Dict[str, Any]:
        """执行任务直到模型给出最终答复或触顶。

        history 为既往对话消息（user/assistant，不含工具细节），用于会话式多轮上下文。
        """
        messages: List[Dict[str, Any]] = []
        if system:
            messages.append({"role": "system", "content": system})
        if history:
            messages.extend(history)
        messages.append({"role": "user", "content": task})

        total_in = 0
        total_out = 0
        last_content = ""

        for round_no in range(1, self.max_rounds + 1):
            self._emit({"type": "round", "round": round_no})
            try:
                resp = await self._invoke_chat(messages)
            except Exception as e:
                self._emit({"type": "final", "round": round_no, "reason": "error",
                            "error": f"{type(e).__name__}: {e}"})
                return {"response": "", "rounds": round_no, "reason": "error",
                        "error": f"{type(e).__name__}: {e}",
                        "prompt_tokens": total_in, "completion_tokens": total_out}

            usage = resp.get("usage") or {}
            total_in += int(usage.get("prompt_tokens", 0))
            total_out += int(usage.get("completion_tokens", 0))
            self._emit({"type": "usage", "prompt_tokens": total_in,
                        "completion_tokens": total_out})

            # 金额护栏（D6 同源）：token 触顶立即停
            if self.token_cap and total_in + total_out > self.token_cap:
                self._emit({"type": "final", "round": round_no, "reason": "token_cap"})
                return {"response": last_content, "rounds": round_no, "reason": "token_cap",
                        "error": f"token 熔断（{total_in + total_out}/{self.token_cap}）",
                        "prompt_tokens": total_in, "completion_tokens": total_out}

            calls = resp.get("tool_calls") or []
            if not calls:
                self._emit({"type": "final", "round": round_no, "reason": "answer"})
                return {"response": resp.get("content") or "", "rounds": round_no,
                        "reason": "answer", "error": None,
                        "prompt_tokens": total_in, "completion_tokens": total_out}

            # 有工具调用：回填 assistant 消息（带原始 tool_calls 结构）+ 逐个执行
            messages.append({
                "role": "assistant",
                "content": resp.get("content") or None,
                "tool_calls": [
                    {
                        "id": c["id"],
                        "type": "function",
                        "function": {"name": c["name"],
                                     "arguments": json.dumps(c["arguments"], ensure_ascii=False)},
                    }
                    for c in calls
                ],
            })
            for c in calls:
                self._emit({"type": "tool_call", "round": round_no,
                            "name": c["name"], "arguments": c["arguments"]})
                # 审批门卫（D9）：按策略挂起等待用户裁决；拒绝/超时折叠为失败结果回喂
                if self.approval_hook is not None:
                    allowed = False
                    try:
                        allowed = await self.approval_hook(c["name"], c["arguments"])
                    except Exception as e:
                        self._emit({"type": "approval_decided", "round": round_no,
                                    "id": "", "allowed": False,
                                    "note": f"审批流程异常: {type(e).__name__}"})
                    if not allowed:
                        denied = {"ok": False,
                                  "error": "用户拒绝执行该操作（如需执行请调整审批策略或换一种做法）"}
                        preview = json.dumps(denied, ensure_ascii=False)
                        self._emit({"type": "tool_result", "round": round_no,
                                    "name": c["name"], "ok": False, "preview": preview})
                        messages.append({"role": "tool", "tool_call_id": c["id"],
                                         "content": preview})
                        continue
                result = await run_tool_async(self.tools, c["name"], c["arguments"])
                preview = json.dumps(result, ensure_ascii=False)
                if len(preview) > self.result_preview_chars:
                    preview = preview[:self.result_preview_chars] + "…(截断)"
                # 重复调用提醒：同一工具同一参数第 N 次起，回喂时附警语促使换策略
                key = (c["name"], json.dumps(c["arguments"], sort_keys=True, ensure_ascii=False))
                seen = self._call_counts.get(key, 0) + 1
                self._call_counts[key] = seen
                if seen >= REPEAT_CALL_WARN_AT:
                    preview += (f"\n[系统提示] 该工具已第 {seen} 次以完全相同的参数调用。"
                                "重复同样的调用不会带来新信息；请改变做法或直接给出最终答复。")
                    self._emit({"type": "repeat_warning", "round": round_no,
                                "name": c["name"], "count": seen})
                self._emit({"type": "tool_result", "round": round_no,
                            "name": c["name"], "ok": bool(result.get("ok")),
                            "preview": preview})
                messages.append({
                    "role": "tool",
                    "tool_call_id": c["id"],
                    "content": preview,
                })
            last_content = resp.get("content") or last_content

        # 轮次耗尽：不再给工具，让模型立即总结（无 tools 的最后一次调用）
        self._emit({"type": "final", "round": self.max_rounds, "reason": "max_rounds"})
        try:
            messages.append({"role": "user",
                             "content": "工具轮次已达上限。请基于以上信息立即给出最终答复，不要再调用工具。"})
            resp = await self.chat_fn(messages, [])
            usage = resp.get("usage") or {}
            total_in += int(usage.get("prompt_tokens", 0))
            total_out += int(usage.get("completion_tokens", 0))
            final_text = resp.get("content") or last_content
        except Exception:
            final_text = last_content
        return {"response": final_text, "rounds": self.max_rounds, "reason": "max_rounds",
                "error": "工具轮次达上限，已强制总结",
                "prompt_tokens": total_in, "completion_tokens": total_out}