sorrow404null
60200260ab
Windows runs 'cmd.exe /d /s /c <command>' as a string command line; Linux uses ['/bin/bash', '-lc', command]. Removes implicit shell=True behavior and aligns the agent system prompt per platform.
866 lines
36 KiB
Python
866 lines
36 KiB
Python
"""
|
||
core/agent/tools.py
|
||
===================
|
||
🌟 pi 工具执行管线的 Python 1:1 移植 + 内置工具
|
||
|
||
对照 pi-main 源码：
|
||
packages/agent/src/agent-loop.ts
|
||
- prepareToolCalls() (行 ~470): 校验参数 → 标记错误（不执行）
|
||
- executeTool() (行 ~520): before 钩子 → 执行 → after 钩子 → 结果定型
|
||
- 并行语义: prepare 串行 → 执行并发(Promise.all) → 结果按原始顺序回写
|
||
packages/agent/src/tools/*.ts (coding-agent 内置工具 read/bash/edit/write)
|
||
|
||
JSON Schema 校验：pi 用 validate-json-schema + ai/src/utils/validation.ts
|
||
（structuredClone → normalizeOptionalNulls → Value.Convert → coerceWithJsonSchema →
|
||
全量错误上报 + 回显收到的参数）；这里实现核心子集
|
||
（type/required/properties/enum/items + 可选字段 null 归一化 + 数字/布尔轻量转换），
|
||
零外部依赖。工具自定义预处理对照 pi 的 tool.prepareArguments（见 edit 的 legacy 兼容）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import queue
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Callable, Dict, List, Optional
|
||
|
||
from .types import (AgentMessage, AgentTool, AgentToolResult, AbortSignal,
|
||
ToolCall, new_id)
|
||
from ..platform_shell import kill_process_tree, popen_flags, shell_command # P1-02
|
||
|
||
try:
|
||
from core.debug_log import debug_log as _dbg_log # 🆕 计时观察日志（线程安全/静默）
|
||
except Exception: # 导入失败也不影响工具执行
|
||
def _dbg_log(msg, tag="APP"):
|
||
pass
|
||
|
||
|
||
# ======================================================================
|
||
# JSON Schema 校验（子集）
|
||
# ======================================================================
|
||
# ======================================================================
|
||
# 参数归一化 + JSON Schema 校验（子集）
|
||
# 对照 pi ai/src/utils/validation.ts:317-347 validateToolArguments：
|
||
# structuredClone → normalizeOptionalNulls → Value.Convert
|
||
# → coerceWithJsonSchema → Check → 【报全部错误 + 回显收到的参数】
|
||
# 这里实现核心子集（type/required/properties/enum/items），零外部依赖。
|
||
# ======================================================================
|
||
_TYPE_MAP = {
|
||
"string": str,
|
||
"integer": int,
|
||
"number": (int, float),
|
||
"boolean": bool,
|
||
"array": list,
|
||
"object": dict,
|
||
}
|
||
|
||
|
||
def _norm_types(schema: Dict[str, Any]) -> List[str]:
|
||
"""schema 声明的类型（兼容 TypeBox 的 type 数组形式）"""
|
||
t = schema.get("type")
|
||
if isinstance(t, list):
|
||
return [x for x in t if isinstance(x, str)]
|
||
return [t] if isinstance(t, str) else []
|
||
|
||
|
||
def _type_ok(value: Any, t: str) -> bool:
|
||
"""类型匹配。🌟 bool 不算 integer/number（python 里 bool 是 int 子类）"""
|
||
if t == "integer":
|
||
return isinstance(value, int) and not isinstance(value, bool)
|
||
if t == "number":
|
||
return isinstance(value, (int, float)) and not isinstance(value, bool)
|
||
if t == "boolean":
|
||
return isinstance(value, bool)
|
||
py = _TYPE_MAP.get(t)
|
||
return isinstance(value, py) if py else True
|
||
|
||
|
||
def _join_path(path: str, key: str) -> str:
|
||
return f"{path}.{key}" if path else key
|
||
|
||
|
||
def normalize_and_coerce(args: Any, schema: Dict[str, Any]) -> Any:
|
||
"""归一化 + 轻量强制转换（对照 pi normalizeOptionalNulls + coerceWithJsonSchema）
|
||
|
||
- 可选字段的显式 null → 删除该键（模型常对「未用到的参数」发 null）
|
||
- 数字字段收到纯数字字符串 → integer/number（"30" → 30）
|
||
- boolean 字段收到 "true"/"false" → 布尔
|
||
返回新对象，不修改入参。
|
||
"""
|
||
if not isinstance(schema, dict) or not isinstance(args, dict):
|
||
return args
|
||
props = schema.get("properties") or {}
|
||
required = set(schema.get("required") or [])
|
||
out: Dict[str, Any] = dict(args)
|
||
for key, val in list(out.items()):
|
||
sub = props.get(key)
|
||
if not isinstance(sub, dict):
|
||
continue
|
||
if val is None:
|
||
if key not in required:
|
||
del out[key]
|
||
continue
|
||
out[key] = _coerce_value(val, sub)
|
||
return out
|
||
|
||
|
||
def _coerce_value(val: Any, sub: Dict[str, Any]) -> Any:
|
||
types = _norm_types(sub)
|
||
if isinstance(val, bool) or val is None:
|
||
return val
|
||
if isinstance(val, str):
|
||
s = val.strip()
|
||
if s and ("integer" in types or "number" in types):
|
||
try:
|
||
f = float(s)
|
||
if f.is_integer():
|
||
return int(f)
|
||
if "number" in types:
|
||
return f
|
||
return val # integer 字段收到 1.5 → 保持原值（交由校验报错）
|
||
except ValueError:
|
||
return val
|
||
if "boolean" in types and s.lower() in ("true", "false"):
|
||
return s.lower() == "true"
|
||
return val
|
||
if isinstance(val, dict):
|
||
return normalize_and_coerce(val, sub)
|
||
if isinstance(val, list):
|
||
items = sub.get("items")
|
||
if isinstance(items, dict):
|
||
return [_coerce_value(v, items) if not isinstance(v, dict)
|
||
else normalize_and_coerce(v, items) for v in val]
|
||
return val
|
||
|
||
|
||
def _type_error(path: str, types: List[str], value: Any) -> str:
|
||
t = types[0] if types else "object"
|
||
if t == "string":
|
||
return f"参数 {path} 应为 string"
|
||
if t in ("integer", "number") and isinstance(value, bool):
|
||
return f"参数 {path} 应为 {t}"
|
||
return f"参数 {path} 类型错误: 期望 {t}"
|
||
|
||
|
||
def _check_value(value: Any, schema: Dict[str, Any], path: str,
|
||
errs: List[str], root: bool = False) -> None:
|
||
"""递归收集【全部】校验错误（对照 pi Errors() 全量上报）"""
|
||
if not isinstance(schema, dict):
|
||
return
|
||
if root and not isinstance(value, dict):
|
||
errs.append(f"参数必须是对象，实际是 {type(value).__name__}")
|
||
return
|
||
types = _norm_types(schema) or (["object"] if root else [])
|
||
if types and not any(_type_ok(value, t) for t in types):
|
||
errs.append(_type_error(path, types, value))
|
||
return # 类型不符 → 后续检查无意义
|
||
if "enum" in schema and value not in schema["enum"]:
|
||
errs.append(f"参数 {path} 取值必须是 {schema['enum']} 之一")
|
||
if isinstance(value, dict):
|
||
props = schema.get("properties") or {}
|
||
for req in schema.get("required") or []:
|
||
if req not in value:
|
||
errs.append(f"缺少必填参数: {req}" if root
|
||
else f"参数 {path} 缺少必填字段 {req}")
|
||
for k, v in value.items():
|
||
if k in props:
|
||
_check_value(v, props[k], _join_path(path, k), errs)
|
||
elif isinstance(value, list):
|
||
items = schema.get("items")
|
||
if isinstance(items, dict):
|
||
for i, item in enumerate(value):
|
||
_check_value(item, items, f"{path}[{i}]", errs)
|
||
|
||
|
||
def validate_json_schema(args: Any, schema: Dict[str, Any]) -> Optional[str]:
|
||
"""返回错误描述（多条以「; 」连接）；合法返回 None。"""
|
||
errs: List[str] = []
|
||
_check_value(args, schema, "", errs, root=True)
|
||
return "; ".join(errs) if errs else None
|
||
|
||
|
||
def _validate_value(value: Any, schema: Dict[str, Any], path: str) -> Optional[str]:
|
||
"""（保留旧签名：返回该节点的首个错误）"""
|
||
errs: List[str] = []
|
||
_check_value(value, schema, path, errs)
|
||
return errs[0] if errs else None
|
||
|
||
|
||
def _brief_json(obj: Any, limit: int = 600) -> str:
|
||
"""参数回显（长内容截断，避免 write 的大 content 撑爆错误消息）"""
|
||
try:
|
||
s = json.dumps(obj, ensure_ascii=False)
|
||
except Exception:
|
||
s = repr(obj)
|
||
return s if len(s) <= limit else s[:limit] + f"…(共 {len(s)} 字符)"
|
||
|
||
|
||
# ======================================================================
|
||
# 准备阶段 —— 对照 agent-loop.ts prepareToolCalls
|
||
# ======================================================================
|
||
@dataclass
|
||
class PreparedToolCall:
|
||
"""一次工具调用的完整准备结果（执行前定型）"""
|
||
tool_call: ToolCall
|
||
tool: Optional[AgentTool]
|
||
args: Dict[str, Any]
|
||
error: str = "" # 准备阶段失败原因（未知工具/参数非法）→ 直接返回错误 toolResult
|
||
|
||
|
||
def prepare_tool_calls(assistant: AgentMessage,
|
||
tools: List[AgentTool]) -> List[PreparedToolCall]:
|
||
tool_map = {t.name: t for t in tools}
|
||
prepared: List[PreparedToolCall] = []
|
||
for tc in assistant.tool_calls:
|
||
tool = tool_map.get(tc.name)
|
||
if tool is None:
|
||
prepared.append(PreparedToolCall(
|
||
tool_call=tc, tool=None, args={},
|
||
error=f"未知工具: {tc.name}",
|
||
))
|
||
continue
|
||
if not isinstance(tc.arguments, dict):
|
||
prepared.append(PreparedToolCall(
|
||
tool_call=tc, tool=tool, args={},
|
||
error="工具参数解析失败（JSON 不完整）",
|
||
))
|
||
continue
|
||
# 🌟 工具自定义参数预处理（对照 pi tool.prepareArguments）
|
||
raw_args = tc.arguments
|
||
if tool.prepare_arguments is not None:
|
||
try:
|
||
raw_args = tool.prepare_arguments(raw_args)
|
||
except Exception as e:
|
||
prepared.append(PreparedToolCall(
|
||
tool_call=tc, tool=tool, args=tc.arguments,
|
||
error=f"参数预处理失败: {e}",
|
||
))
|
||
continue
|
||
if not isinstance(raw_args, dict):
|
||
prepared.append(PreparedToolCall(
|
||
tool_call=tc, tool=tool, args=tc.arguments,
|
||
error="参数预处理返回的不是对象",
|
||
))
|
||
continue
|
||
# 🌟 归一化 + 轻量强制转换（null 可选字段删除 / "30" → 30）
|
||
args = normalize_and_coerce(raw_args, tool.parameters)
|
||
err = validate_json_schema(args, tool.parameters)
|
||
if err:
|
||
prepared.append(PreparedToolCall(
|
||
tool_call=tc, tool=tool, args=args,
|
||
error=f"参数校验失败: {err}；收到的参数: {_brief_json(raw_args)}",
|
||
))
|
||
continue
|
||
prepared.append(PreparedToolCall(tool_call=tc, tool=tool,
|
||
args=args))
|
||
return prepared
|
||
|
||
|
||
# ======================================================================
|
||
# 执行阶段 —— 对照 agent-loop.ts executeTool
|
||
# before 钩子 → 执行 → after 钩子 → 异常兜底
|
||
# ======================================================================
|
||
def execute_tool_call(prepared: PreparedToolCall,
|
||
assistant: AgentMessage,
|
||
config,
|
||
signal: AbortSignal,
|
||
on_update: Optional[Callable[[str], None]],
|
||
on_timer: Optional[Callable[[int, int], None]] = None,
|
||
) -> AgentToolResult:
|
||
"""对照 agent-loop.ts executePreparedToolCall + prepareToolCall 的钩子语义
|
||
🆕 on_timer(elapsed_s, timeout_s)：bash 运行中每秒滴一次（气泡读秒）"""
|
||
tc = prepared.tool_call
|
||
|
||
# 准备阶段已失败 → 直接错误结果（pi 同款：不进入执行）
|
||
if prepared.error:
|
||
return AgentToolResult.text(prepared.error, is_error=True)
|
||
if signal.aborted:
|
||
return AgentToolResult.text("操作已中止 (Operation aborted)", is_error=True)
|
||
|
||
# before 钩子（对照 pi: 可修改 args / 拒绝 block / 请求 terminate）
|
||
if config.before_tool_call:
|
||
try:
|
||
decision = config.before_tool_call(
|
||
{"assistant_message": assistant, "tool_call": tc,
|
||
"args": prepared.args, "context": config.tool_context},
|
||
signal)
|
||
except Exception as e:
|
||
return AgentToolResult.text(f"before 钩子异常: {e}", is_error=True)
|
||
if signal.aborted:
|
||
return AgentToolResult.text("操作已中止 (Operation aborted)",
|
||
is_error=True)
|
||
if decision:
|
||
if decision.get("block"):
|
||
# 🐛 修复：旧代码向 AgentToolResult.text() 传了不存在的 terminate 参数
|
||
# （钩子一旦 block 就抛 TypeError）—— 改为直接构造，保留 terminate 语义
|
||
return AgentToolResult(
|
||
content=[{"type": "text",
|
||
"text": decision.get("reason") or "工具执行被拦截"}],
|
||
is_error=True,
|
||
terminate=bool(decision.get("terminate")))
|
||
if decision.get("args") is not None:
|
||
# 🌟 对照 pi applyBeforeToolDecision：钩子改参后【重新校验】
|
||
new_args = decision["args"]
|
||
if not isinstance(new_args, dict):
|
||
return AgentToolResult.text(
|
||
"before 钩子返回的参数不是对象", is_error=True)
|
||
new_args = normalize_and_coerce(new_args, prepared.tool.parameters)
|
||
_verr = validate_json_schema(new_args, prepared.tool.parameters)
|
||
if _verr:
|
||
return AgentToolResult.text(
|
||
f"before 钩子修改后的参数校验失败: {_verr}；"
|
||
f"收到的参数: {_brief_json(decision['args'])}",
|
||
is_error=True)
|
||
prepared.args = new_args
|
||
|
||
t0 = time.time()
|
||
try:
|
||
# 🆕 秒级滴答回调注入 ctx（不污染调用方的 tool_context 字典）
|
||
_ctx = dict(config.tool_context or {})
|
||
if on_timer is not None:
|
||
_ctx["on_timer"] = on_timer
|
||
result = prepared.tool.execute(
|
||
tc.id, prepared.args, signal, on_update, _ctx,
|
||
)
|
||
if not isinstance(result, AgentToolResult):
|
||
# 宽容处理：工具返回 str 也接受
|
||
result = AgentToolResult.text(str(result))
|
||
except Exception as e:
|
||
result = AgentToolResult.text(f"工具执行异常: {e}", is_error=True)
|
||
|
||
# after 钩子
|
||
if config.after_tool_call:
|
||
try:
|
||
config.after_tool_call(tc, result, result.is_error)
|
||
except Exception:
|
||
pass
|
||
|
||
# 结果定型（对照 pi finalizeToolResult：确保 content 结构合法）
|
||
if not result.content:
|
||
result = AgentToolResult.text("(无输出)")
|
||
result.details = {"duration_ms": int((time.time() - t0) * 1000),
|
||
**(result.details or {} if isinstance(result.details, dict) else {})}
|
||
return result
|
||
|
||
|
||
# ======================================================================
|
||
# 截断保护 —— 对照 agent-loop.ts failToolCallsFromTruncatedMessage
|
||
# 助手消息被 length/aborted 截断时，其工具调用参数可能残缺：
|
||
# 一律替换为错误 toolResult，绝不执行残缺调用。
|
||
# ======================================================================
|
||
def fail_tool_calls_from_truncated_message(assistant: AgentMessage,
|
||
reason: str = "aborted") -> List[AgentMessage]:
|
||
"""为每个工具调用生成错误 toolResult 消息（对照 pi 返回值的组装）"""
|
||
msgs: List[AgentMessage] = []
|
||
for tc in assistant.tool_calls:
|
||
# 对照 pi: "Tool call {name} was not executed: the response hit the
|
||
# output token limit before the arguments were complete..."
|
||
text = (f"工具调用 {tc.name} 未执行（{reason}）："
|
||
f"响应在参数完整前达到输出长度上限，部分参数已被丢弃。"
|
||
f"请用完整参数重试该操作。")
|
||
msgs.append(AgentMessage(
|
||
role="toolResult",
|
||
tool_call_id=tc.id,
|
||
tool_name=tc.name,
|
||
content=text,
|
||
is_error=True,
|
||
))
|
||
return msgs
|
||
|
||
|
||
# ======================================================================
|
||
# 内置工具 —— 对照 pi coding-agent 的 read / bash / write / edit
|
||
# ======================================================================
|
||
_MAX_READ_LINES = 2000
|
||
_MAX_OUTPUT_BYTES = 50 * 1024
|
||
|
||
|
||
def _resolve_path(path: str, ctx: Dict[str, Any]) -> str:
|
||
"""相对路径基于 tool_context 的 cwd（默认项目根）"""
|
||
if os.path.isabs(path):
|
||
return os.path.abspath(path)
|
||
cwd = ctx.get("cwd", os.getcwd())
|
||
return os.path.abspath(os.path.join(cwd, path))
|
||
|
||
|
||
def tool_read(tool_call_id: str, args: Dict[str, Any], signal: AbortSignal,
|
||
on_update, ctx: Dict[str, Any]) -> AgentToolResult:
|
||
"""read: 读取文本文件（支持 offset/limit 行窗口），带行号输出"""
|
||
if not args.get("path"):
|
||
return AgentToolResult.text("path 不能为空", is_error=True)
|
||
path = _resolve_path(args["path"], ctx)
|
||
if not os.path.isfile(path):
|
||
return AgentToolResult.text(f"文件不存在: {path}", is_error=True)
|
||
try:
|
||
offset = max(1, int(args.get("offset", 1)))
|
||
except (TypeError, ValueError):
|
||
return AgentToolResult.text("offset 必须是整数", is_error=True)
|
||
try:
|
||
# 🌟 limit 必须有下限：负数会被 python 负索引语义吃掉
|
||
# （旧版 limit=-5 → lines[0:-5] → 除末尾 5 行外全部返回，与直觉完全相反）
|
||
limit = max(1, min(int(args.get("limit", 2000)), _MAX_READ_LINES))
|
||
except (TypeError, ValueError):
|
||
return AgentToolResult.text("limit 必须是整数", is_error=True)
|
||
try:
|
||
with open(path, "r", encoding="utf-8", errors="replace") as f:
|
||
lines = f.readlines()
|
||
except Exception as e:
|
||
return AgentToolResult.text(f"读取失败: {e}", is_error=True)
|
||
total = len(lines)
|
||
if total == 0:
|
||
return AgentToolResult.text("[文件为空（0 行）]")
|
||
if offset > total:
|
||
return AgentToolResult.text(
|
||
f"[起始行 offset={offset} 超出文件范围，该文件共 {total} 行]")
|
||
chunk = lines[offset - 1: offset - 1 + limit]
|
||
numbered = "".join(
|
||
f"{i + offset:6d}\t{line}" for i, line in enumerate(chunk)
|
||
)
|
||
shown_hi = offset - 1 + len(chunk)
|
||
footer = f"\n[已显示 {offset}–{shown_hi} 行，共 {total} 行]"
|
||
if shown_hi < total:
|
||
footer += f"（还有 {total - shown_hi} 行未显示，用 offset={shown_hi + 1} 继续）"
|
||
out = numbered + footer
|
||
if len(out.encode("utf-8")) > _MAX_OUTPUT_BYTES:
|
||
out = out.encode("utf-8")[:_MAX_OUTPUT_BYTES].decode("utf-8", "ignore")
|
||
out += "\n[输出超过 50KB 已截断]"
|
||
return AgentToolResult.text(out)
|
||
|
||
|
||
def tool_bash(tool_call_id: str, args: Dict[str, Any], signal: AbortSignal,
|
||
on_update, ctx: Dict[str, Any]) -> AgentToolResult:
|
||
"""bash: 执行 shell 命令（默认 120s 超时）。对照 pi: bash 是 sequential 工具
|
||
|
||
🆕 秒级滴答：Popen + communicate(timeout=1) 循环，每秒：
|
||
① 日志 [timer] bash 读秒 N/Ts
|
||
② on_timer(N, T) 推前端气泡读秒
|
||
③ 到期杀进程树（Windows taskkill /T，连孤儿子进程一起杀）→ 真超时
|
||
"""
|
||
if signal.aborted:
|
||
return AgentToolResult.text("操作已中止 (Operation aborted)", is_error=True)
|
||
command = args.get("command", "")
|
||
if not command.strip():
|
||
return AgentToolResult.text("command 不能为空", is_error=True)
|
||
timeout = min(float(args.get("timeout", 120)), 600)
|
||
cwd = ctx.get("cwd") or os.getcwd()
|
||
# P1-02：显式 shell 契约（Windows cmd.exe /d /c；Linux /bin/bash -lc），
|
||
# 不再依赖 shell=True 的平台默认值；ctx 的 "shell" 键已废弃（bash 工具即 shell 工具）
|
||
on_timer = ctx.get("on_timer")
|
||
t0 = time.time()
|
||
# 🆕 计时观察①：计时器启动时刻 + 模型实际传的 timeout 值
|
||
_dbg_log(f"[timer] bash 开始 timeout={timeout:.0f}s "
|
||
f"(显式={args.get('timeout')}) cmd={command[:80]!r}")
|
||
|
||
def _tick(elapsed_i: int):
|
||
"""每秒一次：日志 + 推前端（静默吞异常，绝不影响执行）"""
|
||
_dbg_log(f"[timer] bash 读秒 {elapsed_i}/{int(timeout)}s")
|
||
try:
|
||
if on_timer:
|
||
on_timer(elapsed_i, int(timeout))
|
||
except Exception:
|
||
pass
|
||
|
||
def _kill_tree(proc):
|
||
"""杀整个进程树：Windows taskkill /F /T；Linux 独立进程组
|
||
SIGTERM → 宽限 → SIGKILL（见 core.platform_shell.kill_process_tree）。"""
|
||
kill_process_tree(proc)
|
||
|
||
# 🐛 修复：text=True 不带 encoding 时按系统码页（中文 Windows=GBK）解码，
|
||
# 子进程输出 UTF-8（python/git/中文 echo）→ _readerthread UnicodeDecodeError。
|
||
# 强制 UTF-8 + 容错替换；PYTHONIOENCODING 让 python 子进程也按 UTF-8 输出。
|
||
_env = dict(os.environ, PYTHONIOENCODING="utf-8")
|
||
try:
|
||
proc = subprocess.Popen(
|
||
shell_command(command), # P1-02：显式 shell 命令
|
||
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||
text=True, encoding="utf-8", errors="replace",
|
||
cwd=cwd, env=_env,
|
||
**popen_flags(), # P1-02：Linux 独立进程组（start_new_session）
|
||
)
|
||
except Exception as e:
|
||
return AgentToolResult.text(f"执行失败: {e}", is_error=True)
|
||
|
||
timed_out = False
|
||
out_parts: List[str] = []
|
||
err_parts: List[str] = []
|
||
err_header_fed = False
|
||
|
||
# 🆕 实时输出：stdout/stderr 各起一个 reader 线程 → 队列，
|
||
# 主循环（仍在 worker 线程内）抽干队列并回调 on_update。
|
||
# 绝不从 reader 线程直接回调 —— UI 侧 _on_tool_updated 会改 timeline，跨线程不安全。
|
||
_q: "queue.Queue" = queue.Queue()
|
||
|
||
def _reader(stream, tag):
|
||
try:
|
||
for line in iter(stream.readline, ""):
|
||
_q.put((tag, line))
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
try:
|
||
stream.close()
|
||
except Exception:
|
||
pass
|
||
|
||
_t_out = threading.Thread(target=_reader, args=(proc.stdout, "out"), daemon=True)
|
||
_t_err = threading.Thread(target=_reader, args=(proc.stderr, "err"), daemon=True)
|
||
_t_out.start()
|
||
_t_err.start()
|
||
|
||
def _feed(tag, chunk):
|
||
"""抽到一块输出：累积 + 推实时流（只在 worker 线程内调用）"""
|
||
nonlocal err_header_fed
|
||
piece = chunk
|
||
if tag == "err":
|
||
err_parts.append(chunk)
|
||
if not err_header_fed:
|
||
err_header_fed = True
|
||
piece = "[stderr]\n" + chunk
|
||
else:
|
||
out_parts.append(chunk)
|
||
try:
|
||
if on_update and piece:
|
||
on_update(piece)
|
||
except Exception:
|
||
pass
|
||
|
||
def _drain():
|
||
"""抽干队列里已到达的输出（单轮上限防极端刷屏卡死）"""
|
||
for _ in range(500):
|
||
try:
|
||
tag, chunk = _q.get_nowait()
|
||
except queue.Empty:
|
||
return
|
||
_feed(tag, chunk)
|
||
|
||
last_tick = 0 # 首次滴答仍在 elapsed=1（与旧 communicate 实现一致，不在 0 秒多滴一次）
|
||
try:
|
||
while True:
|
||
# ① 最多 0.2s 醒一次 → 把新输出实时推给前端
|
||
try:
|
||
tag, chunk = _q.get(timeout=0.2)
|
||
_feed(tag, chunk)
|
||
except queue.Empty:
|
||
pass
|
||
_drain()
|
||
|
||
# ② 进程已退出：等 reader 读完管道残余 → 收尾
|
||
if proc.poll() is not None:
|
||
_t_out.join(timeout=1.0)
|
||
_t_err.join(timeout=1.0)
|
||
_drain()
|
||
break
|
||
|
||
# ③ 每秒读秒 + 中止/超时判定（原语义不变）
|
||
elapsed_i = int(time.time() - t0)
|
||
if elapsed_i > last_tick:
|
||
last_tick = elapsed_i
|
||
if signal.aborted:
|
||
_kill_tree(proc)
|
||
_t_out.join(timeout=1.0)
|
||
_t_err.join(timeout=1.0)
|
||
_drain()
|
||
return AgentToolResult.text(
|
||
"操作已中止 (Operation aborted)", is_error=True)
|
||
_tick(elapsed_i)
|
||
if time.time() - t0 >= timeout:
|
||
# 到期 → 真杀（进程树）
|
||
timed_out = True
|
||
_kill_tree(proc)
|
||
try:
|
||
proc.wait(timeout=15)
|
||
except Exception:
|
||
pass
|
||
_t_out.join(timeout=5.0)
|
||
_t_err.join(timeout=5.0)
|
||
_drain()
|
||
break
|
||
except Exception as e:
|
||
try:
|
||
_kill_tree(proc)
|
||
except Exception:
|
||
pass
|
||
return AgentToolResult.text(f"执行失败: {e}", is_error=True)
|
||
|
||
out = "".join(out_parts)
|
||
err = "".join(err_parts)
|
||
code = proc.returncode if proc.returncode is not None else -1
|
||
|
||
dur = time.time() - t0
|
||
if timed_out:
|
||
# 🆕 计时观察②：计时器到期
|
||
_dbg_log(f"[timer] bash 超时触发 设定={timeout:.0f}s "
|
||
f"实际={dur:.1f}s cmd={command[:80]!r}")
|
||
return AgentToolResult.text(f"命令超时（>{timeout:.0f}s）已终止", is_error=True)
|
||
# 🆕 计时观察③：正常结束 + 实际耗时
|
||
_dbg_log(f"[timer] bash 正常结束 dur={dur:.1f}s "
|
||
f"exit={code} 设定timeout={timeout:.0f}s")
|
||
result = f"$ {command}\n"
|
||
if out:
|
||
result += out if out.endswith("\n") else out + "\n"
|
||
if err:
|
||
result += f"[stderr]\n{err}"
|
||
result += f"\n[exit {code}] ({dur:.1f}s)"
|
||
if len(result.encode("utf-8")) > _MAX_OUTPUT_BYTES:
|
||
result = result.encode("utf-8")[:_MAX_OUTPUT_BYTES].decode("utf-8", "ignore")
|
||
result += "\n[输出超过 50KB 已截断]"
|
||
return AgentToolResult.text(result, is_error=code != 0,
|
||
details={"exit_code": code})
|
||
|
||
|
||
def _atomic_write(path: str, content: str) -> None:
|
||
"""原子写：同目录临时文件 + os.replace（避免半截文件）
|
||
|
||
保留原有 newline 语义（默认 None → 平台换行翻译），仅增加原子性。
|
||
"""
|
||
d = os.path.dirname(os.path.abspath(path)) or "."
|
||
fd, tmp = tempfile.mkstemp(dir=d, prefix=".hocode_w_", suffix=".tmp")
|
||
try:
|
||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
os.replace(tmp, path) # Windows/POSIX 均为原子替换
|
||
except BaseException:
|
||
try:
|
||
os.remove(tmp)
|
||
except Exception:
|
||
pass
|
||
raise
|
||
|
||
|
||
def tool_write(tool_call_id: str, args: Dict[str, Any], signal: AbortSignal,
|
||
on_update, ctx: Dict[str, Any]) -> AgentToolResult:
|
||
"""write: 创建/覆盖文件（自动建父目录）"""
|
||
if not args.get("path"):
|
||
return AgentToolResult.text("path 不能为空", is_error=True)
|
||
path = _resolve_path(args["path"], ctx)
|
||
content = args.get("content", "")
|
||
if content is None:
|
||
content = ""
|
||
if not isinstance(content, str):
|
||
content = str(content)
|
||
try:
|
||
parent = os.path.dirname(path)
|
||
if parent:
|
||
os.makedirs(parent, exist_ok=True)
|
||
_atomic_write(path, content)
|
||
except Exception as e:
|
||
return AgentToolResult.text(f"写入失败: {e}", is_error=True)
|
||
return AgentToolResult.text(f"已写入 {len(content)} 字符 → {path}")
|
||
|
||
|
||
def _prepare_edit_args(args: Any) -> Any:
|
||
"""归一化 edit 参数（对照 pi edit.ts:56-70 prepareEditArguments）
|
||
|
||
- edits 为 JSON 字符串 → 解析为数组
|
||
- edits 内条目为 JSON 字符串 → 解析为对象
|
||
- 兼容 legacy 单条形式 {path, oldText, newText} → 包装为 edits:[{...}]
|
||
"""
|
||
if not isinstance(args, dict):
|
||
return args
|
||
out = dict(args)
|
||
edits = out.get("edits")
|
||
if isinstance(edits, str):
|
||
try:
|
||
parsed = json.loads(edits)
|
||
if isinstance(parsed, list):
|
||
out["edits"] = parsed
|
||
elif isinstance(parsed, dict):
|
||
out["edits"] = [parsed]
|
||
except Exception:
|
||
pass # 解析不了就交给校验层报错
|
||
if out.get("edits") is None and ("oldText" in out or "newText" in out):
|
||
ed: Dict[str, Any] = {}
|
||
if "oldText" in out:
|
||
ed["oldText"] = out["oldText"]
|
||
if "newText" in out:
|
||
ed["newText"] = out["newText"]
|
||
out = {"path": out.get("path"), "edits": [ed]}
|
||
if isinstance(out.get("edits"), list):
|
||
norm = []
|
||
for it in out["edits"]:
|
||
if isinstance(it, str):
|
||
try:
|
||
it = json.loads(it)
|
||
except Exception:
|
||
pass
|
||
norm.append(it)
|
||
out["edits"] = norm
|
||
return out
|
||
|
||
|
||
def _plan_edits(content: str, edits: List[Any]):
|
||
"""对【原始内容】定位每条 oldText（要求恰好 1 次）+ 区间重叠检测
|
||
|
||
对照 pi edit-diff.ts:348「edits[i] and edits[j] overlap … Merge them into one edit」。
|
||
返回 (错误信息, [(start, end, index, new_text), ...] 已按 start 排序)
|
||
"""
|
||
spans: List[tuple] = []
|
||
for i, ed in enumerate(edits):
|
||
if not isinstance(ed, dict):
|
||
return f"第 {i + 1} 条 edit 不是对象", None
|
||
old = ed.get("oldText", "")
|
||
if not isinstance(old, str):
|
||
return f"第 {i + 1} 条 edit 的 oldText 必须是字符串", None
|
||
if old == "":
|
||
return f"第 {i + 1} 条 edit 的 oldText 不能为空", None
|
||
c = content.count(old)
|
||
if c == 0:
|
||
return f"第 {i + 1} 条 edit 未找到匹配文本（oldText 不存在或已变化）", None
|
||
if c > 1:
|
||
return f"第 {i + 1} 条 edit 匹配到 {c} 处（要求唯一），请提供更长的上下文", None
|
||
start = content.index(old)
|
||
spans.append((start, start + len(old), i, ed.get("newText", "")))
|
||
ordered = sorted(spans)
|
||
for a, b in zip(ordered, ordered[1:]):
|
||
if b[0] < a[1]:
|
||
return (f"edits[{a[2]}] 与 edits[{b[2]}] 区域重叠，"
|
||
f"请合并为一条 edit 或改为互不相交的修改"), None
|
||
return None, ordered
|
||
|
||
|
||
def tool_edit(tool_call_id: str, args: Dict[str, Any], signal: AbortSignal,
|
||
on_update, ctx: Dict[str, Any]) -> AgentToolResult:
|
||
"""edit: 精确文本替换（edits: [{oldText, newText}]）
|
||
|
||
🌟 语义对齐 pi：全部 edit 都对【原始文件内容】定位，要求各自唯一且区间互不重叠，
|
||
然后按偏移一次性重建（不是逐条 replace 的增量语义）。
|
||
"""
|
||
if not args.get("path"):
|
||
return AgentToolResult.text("path 不能为空", is_error=True)
|
||
path = _resolve_path(args["path"], ctx)
|
||
if not os.path.isfile(path):
|
||
return AgentToolResult.text(f"文件不存在: {path}", is_error=True)
|
||
edits = args.get("edits", [])
|
||
if not isinstance(edits, list) or not edits:
|
||
return AgentToolResult.text("edits 不能为空", is_error=True)
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
content = f.read()
|
||
except Exception as e:
|
||
return AgentToolResult.text(f"读取失败: {e}", is_error=True)
|
||
err, ordered = _plan_edits(content, edits)
|
||
if err:
|
||
return AgentToolResult.text(err, is_error=True)
|
||
# 按偏移从后往前替换（前面的偏移不受影响）
|
||
out = content
|
||
for start, end, _i, new_text in reversed(ordered):
|
||
out = out[:start] + (new_text if isinstance(new_text, str) else str(new_text)) + out[end:]
|
||
try:
|
||
_atomic_write(path, out)
|
||
except Exception as e:
|
||
return AgentToolResult.text(f"编辑失败: {e}", is_error=True)
|
||
return AgentToolResult.text(f"已应用 {len(ordered)} 处编辑 → {path}")
|
||
|
||
|
||
# ======================================================================
|
||
# 默认工具集 —— 对照 pi 默认启用 read/bash/edit/write
|
||
# ======================================================================
|
||
|
||
_TEXT_TOOL_RE = re.compile(r"<(bash|read)>(.*?)</\1>", re.DOTALL)
|
||
|
||
|
||
def parse_text_tool_calls(content: str):
|
||
"""
|
||
兜底解析（haocode 扩展，pi 无此层）：
|
||
对不支持 tools API 的供应商/模型——它们会把工具调用用纯文字"演"出来
|
||
（例如 <bash>ls</bash>、<read>路径</read>），本函数识别单参数工具
|
||
bash / read 并转成真 ToolCall 供循环执行。
|
||
write / edit 参数多、文本歧义大，不做兜底（保持安全）。
|
||
返回 (原文, [ToolCall])；未命中时 calls 为空列表。
|
||
"""
|
||
if not content or "<" not in content:
|
||
return content, []
|
||
calls: List[ToolCall] = []
|
||
|
||
def _sub(m):
|
||
name, payload = m.group(1), m.group(2).strip()
|
||
if not payload:
|
||
return m.group(0)
|
||
if name == "bash":
|
||
calls.append(ToolCall(id=new_id("txtcall"), name="bash",
|
||
arguments={"command": payload}))
|
||
elif name == "read":
|
||
calls.append(ToolCall(id=new_id("txtcall"), name="read",
|
||
arguments={"path": payload}))
|
||
return m.group(0) # 保留原文（UI 已渲染，不回改；执行由 tool_calls 驱动）
|
||
|
||
cleaned = _TEXT_TOOL_RE.sub(_sub, content)
|
||
return cleaned, calls
|
||
|
||
def default_tools() -> List[AgentTool]:
|
||
return [
|
||
AgentTool(
|
||
name="read", label="读取文件",
|
||
description="读取文本文件内容（带行号）。支持 offset/limit 按行窗口读取大文件。",
|
||
parameters={
|
||
"type": "object",
|
||
"properties": {
|
||
"path": {"type": "string", "description": "文件路径"},
|
||
"offset": {"type": "integer", "description": "起始行号（从 1 开始）"},
|
||
"limit": {"type": "integer", "description": "最多读取行数（默认 2000）"},
|
||
},
|
||
"required": ["path"],
|
||
},
|
||
execute=tool_read,
|
||
),
|
||
AgentTool(
|
||
name="bash", label="执行命令",
|
||
description="执行 shell 命令并返回 stdout/stderr/退出码。默认 120 秒超时。",
|
||
parameters={
|
||
"type": "object",
|
||
"properties": {
|
||
"command": {"type": "string", "description": "要执行的命令"},
|
||
"timeout": {"type": "number", "description": "超时秒数（默认 120，最大 600）"},
|
||
},
|
||
"required": ["command"],
|
||
},
|
||
execute=tool_bash,
|
||
execution_mode="sequential", # 对照 pi: bash 整批串行
|
||
),
|
||
AgentTool(
|
||
name="write", label="写入文件",
|
||
description="创建或覆盖写入文件（自动创建父目录）。",
|
||
parameters={
|
||
"type": "object",
|
||
"properties": {
|
||
"path": {"type": "string", "description": "文件路径"},
|
||
"content": {"type": "string", "description": "文件内容"},
|
||
},
|
||
"required": ["path", "content"],
|
||
},
|
||
execute=tool_write,
|
||
),
|
||
AgentTool(
|
||
name="edit", label="编辑文件",
|
||
description=("对文件做精确文本替换。edits 中每条 oldText 必须在原文件中唯一，"
|
||
"且各条区间不得重叠（重叠请合并为一条）。"),
|
||
parameters={
|
||
"type": "object",
|
||
"properties": {
|
||
"path": {"type": "string", "description": "文件路径"},
|
||
"edits": {
|
||
"type": "array",
|
||
"items": {
|
||
"type": "object",
|
||
"properties": {
|
||
"oldText": {"type": "string"},
|
||
"newText": {"type": "string"},
|
||
},
|
||
"required": ["oldText"],
|
||
},
|
||
"description": "替换操作列表",
|
||
},
|
||
},
|
||
"required": ["path", "edits"],
|
||
},
|
||
execute=tool_edit,
|
||
prepare_arguments=_prepare_edit_args,
|
||
),
|
||
]
|