""" pipeline-service v2: Agent Executor — 智能执行引擎 对照 Hermes Agent 的 run_conversation() + tool dispatch： - 多轮 tool-calling loop，max_turns 可配置 - Memory + Skills 注入每轮 system prompt - 上下文压缩（近 token 上限时自动压缩） - 并行 tool calls（同一轮内独立工具同时执行） - 消息数组污染防护（tool_call JSON 不进入 LLM 上下文） - 会话持久化（用户+项目隔离） - ask_user 澄清机制 - delegate_subtask 子任务分发 使用方法： executor = AgentExecutor(config, project_id=..., user_id=...) async for chunk in executor.run("用户输入"): yield chunk # NDJSON 流式输出 """ import asyncio import json import logging import os import time from typing import AsyncGenerator, Dict, List, Optional logger = logging.getLogger("pipeline.agent_executor") # 项目管理工具别名表：LLM 常编造近义名（停止/终止/删除项目的各种说法）， # 归一到真实工具名。只加「语义等价」的别名，不扩大权限（权限校验在代码层）。 # 注意：不能映射既有真实工具名（如 list_projects 是 sdlc_ability 的产线项目列表 # 工具，语义与 list_my_projects 不同），否则会劫持既有功能。 _PROJECT_TOOL_ALIASES = { "stop_project": "pause_project", "terminate_project": "pause_project", "suspend_project": "pause_project", "halt_project": "pause_project", "unpause_project": "resume_project", "continue_project": "resume_project", "remove_project": "delete_project", "destroy_project": "delete_project", "get_project_info": "project_info", "show_project": "project_info", "my_projects": "list_my_projects", } from .workspace import WORKSPACE_BASE, GENERAL_SPACE # 产线/项目管理类工具（category=='project'）。产线隔离铁律：纯通用会话 # （无产线插件）绝不挂载——否则通用助手能看到并操作其他产线的项目 # （2026-09-05 修复）。过滤同时作用于：① schema（LLM 看不见） # ② 执行层（看见也拒绝），纵深防御。 _PROJECT_TOOL_NAMES = { "switch_project", "create_project", "project_info", "list_my_projects", "pause_project", "resume_project", "delete_project", } # 产线耦合工具拦截（2026-09-08 更新）：run_command 已从 generic 拒绝名单移除—— # strict bwrap 档（不挂平台目录、可写根=用户 _general/{uid}）根治了「shell 枚举 # 全工作空间」的泄漏路径，_t_run_command 对 generic 强制 strict=True，无 bwrap 拒绝。 # 名单保留机制本身，未来若有无法沙箱化的工具仍可加入。 _GENERIC_DENIED_TOOLS = set() # 用户敏感信息库工具名（2026-09-17）：与 pipeline_core.agent_config.GENERAL_TOOLS 里 # category="secret" 的工具一一对应，与 secret_vault.SECRET_HANDLERS 的键一致。 # 三处同步：工具定义(core) + handlers 映射(本文件 _dispatch_sdlc_tool) + handler 实现(secret_vault)。 _SECRET_TOOL_NAMES = ( "list_secrets", "save_secret", "use_secret", "delete_secret", "set_secret_status", "detect_secret", ) # ── 默认工具定义（在 pipeline-core 未加载时使用）── _BUILTIN_ASK_USER_SCHEMA = { "name": "ask_user", "description": "向用户提问澄清意图。不确定用户想要什么时使用。", "parameters": { "type": "object", "properties": { "question": {"type": "string", "description": "要问的问题"}, }, "required": ["question"], }, } class AgentExecutor: """Agent 执行引擎。 负责： 1. 组装 system prompt（配置 + 记忆 + 技能 + 工具） 2. 运行 tool-loop（LLM 决策 → 工具执行 → 回传结果） 3. 上下文压缩 4. 会话持久化 """ # 收尾预留轮次（2026-09-11）：轮次预算剩这么多轮时注入「立即收尾」指令， # 强制 LLM 停止调工具、基于已有材料输出最终结论。根因见 run() 内 5-pre0 段注释 #（v1 有 _FORCE_PRODUCE_HINT、v2 此前缺等价机制 → 长任务撞穿 max_turns 后 # 静默退出，用户只拿到中间进度句）。 _WRAP_UP_RESERVE = 3 # 辅助调用（压缩摘要/技能选择/意图分类）超时上限（2026-09-11）。 # 这些是「可降级」调用：失败有兜底（摘要退化为截断文本、技能选择退化为目录层）。 # 给短超时让它们快速失败走兜底，而不是占用端点默认 300s × 3 重试把整轮卡死 #（实测上游抖动窗口内 utility 调用连烧 15 分钟才放弃，会话全程无响应）。 _UTILITY_TIMEOUT = 60 # 续跑门禁（2026-09-11 商机产线实测）：模型输出纯文本「意图句」就结束 turn、 # 不调工具 → 循环误判为最终答案终止会话，用户拿到半截话（实测连续 4 个会话 # 在 tool_call_count 7~11 时这样夭折，报告/总结从未产出）。 # 门禁条件：已有工作推进(tool_call_count>0) 且回复短于 _CONTINUE_GATE_MIN_CHARS #（真结论/报告远长于此）且催促未达上限 → 注入续跑指令继续循环； # 达上限后尊重模型终止（防死循环）。 _CONTINUE_GATE_MIN_CHARS = 400 _CONTINUE_GATE_MAX = 2 # 主调用单次超时（2026-09-11）：端点缺省 330s 太长——上游超时窗内一次失败 # 要烧 330s×3 重试。实测最长成功响应 3212 token <60s，180s 留 3 倍余量足够， # 失败快速返回交给会话级重试跨越窗口。 _MAIN_TIMEOUT = 180 # 会话级重试次数（含首次共 3 次尝试）+ 退避秒数：上游超时窗实测分钟级、 # 间歇出现；3 次尝试 × 180s + 2×15s 退避 ≈ 9.5 分钟跨度，可跨越多数窗口。 _MAIN_ATTEMPTS = 3 _MAIN_RETRY_BACKOFF = 15 def __init__( self, config, # AgentConfig (from pipeline_core) project_id: str = "", user_id: str = "", workspace_dir: str = "", model_name: str = None, role: str = "", # 产线内角色（可选，如 develop/design；驾驶舱 agent 为空） base_url: str = "", # 请求 base_url（scheme://host/），供 slash 命令生成 widget 绝对 URL generic: bool = False, # True = 纯通用会话（不解析项目、不挂产线能力） session=None, # GatewaySession（会话级状态：approve_all/pending_confirm） session_id: str = "", # 会话内唯一标识（web 多 tab 独立会话；历史隔离键） default_pipeline_id: str = "", # 无当前项目时的默认产线（入口指定，如 bidding_general） delegate_depth: int = 0, # 委派深度（0=顶层会话；子agent=1，禁止再委派） pii_session_key: str = "", # PII ephemeral 缓存键（gateway _session_key；空=无 PII 层） ): self.config = config self.project_id = project_id self.user_id = user_id self.pii_session_key = pii_session_key self.org_id = "" # loaded from project context self.pipeline_id = default_pipeline_id or "" # 默认产线（入口指定）；有项目时被项目 pipeline_id 覆盖 self.role = role # 产线内角色（技能/工具/prompt 按角色加载） self.generic = generic # 是否纯通用会话 self._session = session # GatewaySession（approve_all/pending_confirm 会话级状态） self.session_id = session_id # 会话内唯一标识（历史消息按此隔离） self.space = GENERAL_SPACE # 项目空间键：generic→'general'，产线→真实 pipeline_id self.workspace_dir = workspace_dir or WORKSPACE_BASE self.model_name = model_name or config.model_name self.base_url = base_url # 运行状态 self._msgs: List[dict] = [] # LLM 消息数组 self._turn_count: int = 0 self._tool_call_count: int = 0 self._auto_push_count: int = 0 # 限制 auto-inject 次数 self._session_id: str = "" self._started_at: float = 0.0 self._todos: List[dict] = [] # 会话内任务清单 [{done, content}] # ── 委派运行时（2026-09-10）── self._delegate_depth: int = delegate_depth # 子agent不能再委派（subagents.MAX_DEPTH） self._subagent_id: str = "" # 本 executor 若是后台子agent，其 id（心跳/steer/stop 钩子用） # ── 五级作用域工具解析结果（2026-09-10）── self._has_images: bool = False # 本轮消息含图片 parts（LLM 失败降级重试用） self._allowed_tools: Optional[set] = None # None=未解析(回退旧行为)；set=执行层门禁允许集 self._capability_tool_names: set = set() # 技能 frontmatter 声明的 capability 工具名 self._tool_scope_trace: List[str] = [] # 诊断轨迹（/tools slash 可展示） # 懒加载 self._tool_registry = None self._skill_loader = None self._memory_store = None self._compressor = None self._session_mgr = None # ═══════════════════════════════════════════════════════ # 主入口 # ═══════════════════════════════════════════════════════ async def run(self, user_input: str, history: List[dict] = None, image_parts: List[dict] = None) -> AsyncGenerator[str, None]: """执行 agent 主循环。 Args: user_input: 用户输入 history: 历史消息列表 [{"role":"user"|"assistant","content":"..."}] image_parts: 上传图片的 OpenAI 多模态 parts（upload_tools.build_image_parts 产物）。非空时首条用户消息构造为多模态 content 数组 [{"type":"text",...}, {"type":"image_url",...}]—— llm_bridge → llm_v1 端点 → inference 全链原样透传 messages， 视觉模型直接看图；非视觉模型由 _call_llm 的降级重试兜底。 Yields: NDJSON 行：{"type":"progress"|"tool_call"|"reply"|"error", ...} """ self._started_at = time.time() self._turn_count = 0 self._tool_call_count = 0 self._continue_nudges = 0 # Step 0: 初始化（解析 pipeline_id / skill_loader 等，slash 命令也需要） await self._init_components() # Step 0.5: slash 命令短路（/xxx 直接执行，不进入 LLM 循环） if user_input and user_input.strip().startswith("/"): from pipeline_core import resolve_slash_command, parse_slash_args cmd = resolve_slash_command(user_input.strip(), self.pipeline_id, self.role) if cmd and cmd.handler: args = parse_slash_args(user_input.strip()) try: result = await cmd.handler(args, self._build_slash_ctx()) except Exception as e: logger.error(f"slash /{cmd.name} error: {e}") result = f"ERROR: {str(e)[:300]}" yield json.dumps({"type": "reply", "message": result}, ensure_ascii=False) + "\n" return # 未知 slash 命令：提示并列出可用命令 from pipeline_core import get_slash_commands visible = get_slash_commands(self.pipeline_id, self.role) help_text = "未知命令。可用: " + ", ".join(f"/{n}" for n in sorted(visible)) yield json.dumps({"type": "reply", "message": help_text}, ensure_ascii=False) + "\n" return # Step 2: 加载历史 hist_msgs = await self._load_history(history) # Step 3: 组装 system prompt system_prompt = await self._build_system_prompt(user_input) # Step 4: 构建消息数组 self._msgs = [{"role": "system", "content": system_prompt}] if hist_msgs: self._msgs.extend(hist_msgs) if image_parts: # 原生视觉（2026-09-10）：OpenAI 多模态 content 数组。 # 文本部分放最前，图片 parts 依次跟随；_has_images 标记供降级重试用。 self._msgs.append({"role": "user", "content": [{"type": "text", "text": user_input}] + list(image_parts)}) self._has_images = True else: self._msgs.append({"role": "user", "content": user_input}) # Step 4.5: 危险命令确认回复短路（pending_confirm + 确认/全部确认/取消） confirm_decision = self._detect_confirm_decision(user_input) if confirm_decision and self._session and self._session.pending_confirm: pending = self._session.pending_confirm if confirm_decision == "cancel": self._session.pending_confirm = None yield json.dumps({"type": "reply", "message": "已取消该操作。"}, ensure_ascii=False) + "\n" await self._save_turn(user_input, "已取消该操作。") return if confirm_decision == "approve_all": self._session.approve_all = True self._session.pending_confirm = None pending_tool = pending.get("tool", "") pending_params = pending.get("params") or {} # 回填 assistant tool_call → 执行 → 回填 tool result，随后继续 Tool Loop fake_id = f"call_{int(time.time() * 1000)}" self._msgs.append({ "role": "assistant", "content": None, "tool_calls": [{ "id": fake_id, "type": "function", "function": {"name": pending_tool, "arguments": json.dumps(pending_params, ensure_ascii=False)}, }], }) yield json.dumps({"type": "tool_call", "tool": pending_tool, "params": pending_params}, ensure_ascii=False) + "\n" result = await self._execute_tool(pending_tool, pending_params) if not result.startswith("未知工具") and not result.startswith("ERROR"): self._tool_call_count += 1 # 显示层上限（2026-09-15 用户要求完整显示，替代旧硬截 [:500]）： # 默认 50000 字符 = 正常工具输出完整显示；超限显式告知（绝不静默截断）。 from .result_cap import cap_tool_result, cap_display_result, resolve_max_chars, resolve_display_max_chars yield json.dumps({"type": "tool_result", "tool": pending_tool, "result": cap_display_result(result, await resolve_display_max_chars())}, ensure_ascii=False) + "\n" # 回填上限（2026-09-11）：工具结果全文回填会撑爆上下文（商机产线实测 # 8 次数据工具 ~78K token → 压缩摘要自身超时 → 会话零产出）。 # 截断显式告知 + 给缩小范围指引，让 LLM 自己改策略。 _capped, _ = cap_tool_result(result, await resolve_max_chars()) self._msgs.append({"role": "tool", "tool_call_id": fake_id, "content": _capped}) # 不 return，继续进入 Tool Loop，让 LLM 基于工具结果继续后续步骤 # Step 5: Tool Loop yield json.dumps({"type": "progress", "message": "思考中..."}, ensure_ascii=False) + "\n" final_reply = "" max_turns = self.config.max_turns for turn in range(max_turns): self._turn_count = turn + 1 # 5-pre0. 收尾门禁（2026-09-11 商机产线实测教训）：轮次预算接近耗尽时 # 注入「立即收尾」指令，强制 LLM 用已有材料给出最终结论。 # 根因：v1 角色 agent 有 _FORCE_PRODUCE_HINT 打断探索循环，v2 会话 agent # 此前没有等价机制——多工具长任务（如商机调研→功能点估算→成本→总结） # 会跑满 max_turns 后静默退出，用户只看到中间进度句（「功能点已算出…」）， # 拿不到要求的总结。收尾轮不再允许调工具，只准产出最终回答。 _wrap_turn = max_turns - self._WRAP_UP_RESERVE if turn == _wrap_turn: self._msgs.append({ "role": "user", "content": ( "⚠️【收尾指令】你的对话轮次即将用尽（仅剩 %d 轮）。" "立即停止调用任何工具，基于已经获取的材料直接给出**完整的最终回答**：" "把用户要求的每一项都答到（数据/结论/建议/总结），" "信息不全的部分如实标注「数据未覆盖」并给出你的专业判断，" "不要再做探索、核对或补数。现在就输出最终结论。" % (max_turns - turn) ), }) logger.info(f"wrap-up hint injected at turn {turn+1}/{max_turns}") # 5-pre. 子 agent 运行时钩子（2026-09-10）：心跳 + steer 消费 + stop 检测。 # 后台委派的子 executor 带 _subagent_id，每轮 tool-loop 边界： # - heartbeat 刷新 meta.updated_at（stale 判活依据） # - steer.txt 有内容 → 作为带外父指令注入消息（消费后删除） # - stop.flag 存在 → 优雅停止（已有部分结果由 subagents._runner 保留） if self._subagent_id: try: from . import subagents as _sa steer_msgs, stop_req = _sa.heartbeat( self.workspace_dir, self._subagent_id) except Exception as e: logger.warning(f"subagent hook failed: {e}") steer_msgs, stop_req = [], False if stop_req: msg = "（子任务被父agent终止，已保留部分结果）" yield json.dumps({"type": "reply", "message": msg}, ensure_ascii=False) + "\n" return for sm in steer_msgs: self._msgs.append({ "role": "user", "content": ("[带外指令——来自父agent的中途纠正，与用户消息同等效力，" f"立即调整方向] {sm}"), }) # 5a. 上下文压缩 if self.config.compression.enabled: await self._maybe_compress() # 5b. LLM 调用（native function calling，返回 dict） # 模型不可用等配置错误必须真实报给用户（error 事件）， # 绝不能让异常打断流——前端会停在"思考中..."干等。 # 会话级重试（2026-09-11）：上游瞬时抖动（TimeoutError/ClientError）时 # 重试一次再放弃。实测上游存在分钟级抖动窗口（同模型同上下文直调成功、 # 产线调用 TimeoutError 3 连败），单次失败即终止会话代价过高（整轮工作丢失）。 # 重试前向用户透出进度，避免"卡死"观感。 resp = None _last_err = None for _attempt in range(self._MAIN_ATTEMPTS): try: resp = await self._call_llm() break except Exception as e: _last_err = e _transient = any(k in type(e).__name__ or k in str(e) for k in ("Timeout", "ClientError", "ServerDisconnected", "ConnectionReset")) if not _transient or _attempt >= self._MAIN_ATTEMPTS - 1: break logger.warning(f"run: LLM 瞬时失败 turn={turn+1} 重试{_attempt+1}: {e}") yield json.dumps({ "type": "progress", "message": "模型上游瞬时超时，正在重试…\n", }, ensure_ascii=False) + "\n" await asyncio.sleep(self._MAIN_RETRY_BACKOFF) if resp is None: logger.error(f"run: LLM 调用失败 turn={turn+1}: {_last_err}") yield json.dumps({"type": "error", "message": str(_last_err)}, ensure_ascii=False) + "\n" return # 5c. 原生 function calling：优先处理 tool_calls native_calls = (resp.get("tool_calls") or []) if isinstance(resp, dict) else [] if native_calls: # 回填 assistant 消息（含 tool_calls，OpenAI 原生格式要求） self._msgs.append({ "role": "assistant", "content": resp.get("content") or None, "tool_calls": native_calls, }) for tc in native_calls: fn = tc.get("function", {}) if isinstance(tc, dict) else {} tool_name = fn.get("name", "") try: tool_params = json.loads(fn.get("arguments") or "{}") except Exception: tool_params = {} # 需要确认的工具：记录待确认并暂停，等待用户确认/全部确认/取消 if self._needs_confirmation(tool_name, tool_params): if self._session: self._session.pending_confirm = {"tool": tool_name, "params": tool_params} yield json.dumps({ "type": "confirm", "tool": tool_name, "params": tool_params, }, ensure_ascii=False) + "\n" await self._save_turn(user_input, f"[需确认] {tool_name}") return yield json.dumps({ "type": "tool_call", "tool": tool_name, "params": tool_params, }, ensure_ascii=False) + "\n" result = await self._execute_tool(tool_name, tool_params) if not result.startswith("未知工具") and not result.startswith("ERROR"): self._tool_call_count += 1 # 完备性/澄清门禁硬拦截（2026-09-08 用户定夺：一C+二A）： # 工具返回 QUESTION: 前缀 = 平台判定「缺完成任务的必备输入」， # 已停在花钱动作之前（上游未调用、零费用）。必须把它当 ask_user # 直接抛给用户等回答，禁止回填后靠 LLM 自觉转达（deepseek 类 # 模型会无视、继续空参硬试烧钱）。ask_user 工具同走此前缀，语义统一。 if isinstance(result, str) and result.startswith("QUESTION:"): question = result[len("QUESTION:"):].strip() yield json.dumps({ "type": "ask_user", "message": question, }, ensure_ascii=False) + "\n" await self._save_turn(user_input, f"[提问] {question}") return # 显示层上限（同 Step 4.5，默认 50000 = 完整显示；超限显式告知） from .result_cap import cap_display_result, resolve_display_max_chars yield json.dumps({ "type": "tool_result", "tool": tool_name, "result": cap_display_result(result, await resolve_display_max_chars()), }, ensure_ascii=False) + "\n" # 原生回填：role=tool + tool_call_id # 回填上限：防数据类工具全文回填撑爆上下文（见 result_cap 模块 docstring） from .result_cap import cap_tool_result, resolve_max_chars _capped, _trunc = cap_tool_result(result, await resolve_max_chars()) if _trunc: logger.info(f"tool_result truncated before refill: {tool_name}") self._msgs.append({ "role": "tool", "tool_call_id": tc.get("id", "") if isinstance(tc, dict) else "", "content": _capped, }) continue # 5d. 文本解析（回退路径，纯文本模型） raw = resp.get("content", "") if isinstance(resp, dict) else str(resp) actions = self._parse_actions(raw) # 5e. 分发处理 for act in actions: action_type = act.get("action", "") if action_type == "reply": # reply 是合法终止动作，直接输出，不再强制注入工具。 # （历史版本会在 tool_call_count==0 时用硬编码关键词强制路由， # 那会剥夺 LLM 诚实降级/能力自省的能力，把 agent 变成路由器。） final_reply = act.get("message", "") # 续跑门禁（2026-09-11）：已有工作推进但回复是短意图句 #（"我再拉一次数据…"）= 模型提前结束 turn 而非真结论。 # 注入续跑指令继续循环，不让会话夭折在半截话上。 if (self._tool_call_count > 0 and len(final_reply) < self._CONTINUE_GATE_MIN_CHARS and self._continue_nudges < self._CONTINUE_GATE_MAX): self._continue_nudges += 1 logger.warning( f"run: 续跑门禁 turn={turn+1} 短回复({len(final_reply)}字)" f" tool_calls={self._tool_call_count} 催促{self._continue_nudges}次") yield json.dumps({ "type": "progress", "message": "检测到未完成的中途回复，继续推进任务…\n", }, ensure_ascii=False) + "\n" self._msgs.append({"role": "assistant", "content": final_reply}) self._msgs.append({"role": "user", "content": ( "【续跑指令】你刚才的回复是中途进度句，不是最终答案，" "任务尚未完成（用户要求的产出还没交付）。" "请继续执行：要么调用工具推进下一步，" "要么直接输出完整的最终结论/报告（含全部要求的分析项）。" "不要只说下一步打算做什么。")}) continue yield json.dumps({ "type": "reply", "message": final_reply, }, ensure_ascii=False) + "\n" await self._save_turn(user_input, final_reply) return elif action_type == "ask_user": # ask_user 是合法终止动作：把问题抛给用户，等待回答。 question = act.get("question", "") yield json.dumps({ "type": "ask_user", "message": question, }, ensure_ascii=False) + "\n" await self._save_turn(user_input, f"[提问] {question}") return elif action_type == "tool_call": tool_name = act.get("tool", "") tool_params = act.get("params", {}) # 需要确认的工具：记录待确认并暂停，等待用户确认/全部确认/取消 if self._needs_confirmation(tool_name, tool_params): if self._session: self._session.pending_confirm = {"tool": tool_name, "params": tool_params} yield json.dumps({ "type": "confirm", "tool": tool_name, "params": tool_params, }, ensure_ascii=False) + "\n" await self._save_turn(user_input, f"[需确认] {tool_name}") return # 执行工具 yield json.dumps({ "type": "tool_call", "tool": tool_name, "params": tool_params, }, ensure_ascii=False) + "\n" result = await self._execute_tool(tool_name, tool_params) if not result.startswith("未知工具") and not result.startswith("ERROR"): self._tool_call_count += 1 # 完备性/澄清门禁硬拦截（同 native 路径，2026-09-08 一C+二A） if isinstance(result, str) and result.startswith("QUESTION:"): question = result[len("QUESTION:"):].strip() yield json.dumps({ "type": "ask_user", "message": question, }, ensure_ascii=False) + "\n" await self._save_turn(user_input, f"[提问] {question}") return # 显示层上限（同 native 路径，默认 50000 = 完整显示；超限显式告知） from .result_cap import cap_display_result, resolve_display_max_chars yield json.dumps({ "type": "tool_result", "tool": tool_name, "result": cap_display_result(result, await resolve_display_max_chars()), }, ensure_ascii=False) + "\n" # 反馈给 LLM（关键：不存原始 tool_call JSON） self._msgs.append({ "role": "assistant", "content": f"已调用 {tool_name}", }) self._msgs.append({ "role": "user", "content": f"工具 {tool_name} 结果：\n{result}", }) elif action_type == "error": final_reply = act.get("message", "处理出错") yield json.dumps({ "type": "error", "message": final_reply, }, ensure_ascii=False) + "\n" return # 超过 max_turns final_reply = final_reply or "任务过于复杂，请简化描述后重试。" yield json.dumps({ "type": "reply", "message": final_reply, }, ensure_ascii=False) + "\n" await self._save_turn(user_input, final_reply) # ═══════════════════════════════════════════════════════ # 初始化 # ═══════════════════════════════════════════════════════ async def _init_components(self): """懒加载各组件""" # 1. 加载 org_id（用户机构优先，fallback 项目机构）+ workspace_dir + pipeline_id # 个人选择 llm 按「用户机构」隔离（default_llm_id 指向本机构 llm），故 org_id 须取 users.orgid； # 项目机构仅作 fallback（无 user_id 的测试场景）。 if self.user_id: try: from sqlor.dbpools import DBPools db = DBPools() async with db.sqlorContext("pipeline") as sor: recs = await sor.sqlExe( "SELECT orgid FROM users WHERE id=${uid}$", {"uid": self.user_id}) if recs: self.org_id = getattr(recs[0], 'orgid', '') or '' except Exception: pass if self.project_id: try: from sqlor.dbpools import DBPools db = DBPools() async with db.sqlorContext("pipeline") as sor: recs = await sor.sqlExe( "SELECT org_id, workspace_dir, pipeline_id FROM sd_projects WHERE id=${pid}$", {"pid": self.project_id}) if recs: if not self.org_id: self.org_id = getattr(recs[0], 'org_id', '') or '' ws = getattr(recs[0], 'workspace_dir', '') or '' if ws: self.workspace_dir = ws self.pipeline_id = getattr(recs[0], 'pipeline_id', '') or '' if not ws: # 存量项目无 workspace_dir：兜底解析项目根目录（与上传落盘同一函数， # 保证「上传放的位置 = agent 读的位置」） from .workspace import get_project_dir_by_id _pdir, _ = await get_project_dir_by_id(sor, self.project_id) if _pdir: self.workspace_dir = _pdir except Exception: pass # 2. Tool Registry # 产线隔离（2026-09-05）：必须每个 executor 独立实例，禁止全局单例。 # 全局单例会让先启动的产线会话注册的工具（如 sdlc 的 list_projects） # 永久残留，之后通用会话把全部工具塞给 LLM → 通用助手看到并调用 # 其他产线的工具。config.tools 已由 load_agent_config 按会话裁剪， # 这里只注册本会话自己的工具。 try: from pipeline_core.tool_registry import ToolRegistry self._tool_registry = ToolRegistry() if self.config.tools: for t in self.config.tools: self._tool_registry.register(t) except ImportError: self._tool_registry = None # 3. Skill Loader — base_dir 动态解析到机构工作目录（多租户隔离） try: from pipeline_core.skill_loader import get_skill_loader base_dir = await self._resolve_skills_base_dir() self._skill_loader = get_skill_loader(base_dir) if base_dir: self._skill_loader.reload() except Exception: self._skill_loader = None # 4. Memory Store try: from pipeline_core.memory_store import get_memory_store self._memory_store = get_memory_store() except ImportError: self._memory_store = None # 5. pipeline_id 为空时 fallback 到默认产线（保证 slash/ability/skill 统一按产线挂载） # 入口已指定 default_pipeline_id 时优先用它（投标/开发产线独立入口），否则用引擎默认。 # 产线隔离（2026-09-05）：纯通用会话（generic）绝不回退默认产线—— # 回退会让 _execute_ability_tool 拿到 sdlc_general 能力包，通用助手 # 就能调用 list_projects 等产线工具。 if not self.pipeline_id and not self.generic: try: from pipeline_core import DEFAULT_ABILITY_ID self.pipeline_id = DEFAULT_ABILITY_ID except ImportError: pass # 6. 项目空间键：generic→'general'（通用助手），否则用真实 pipeline_id（产线之间隔离） self.space = GENERAL_SPACE if self.generic else (self.pipeline_id or GENERAL_SPACE) # 7. 通用会话专属工作目录（2026-09-05 第 6 层泄漏修复）：默认文件根是 # WORKSPACE_BASE 总根（{base}/{org}/{产线}/… 全量可见），LLM 用 # list_files/read_file 就能从目录名枚举所有产线的项目。通用会话改用 # 用户专属目录 _general/{user_id}，_resolve_ws_path 越界保护把它圈死。 if self.generic: from .workspace import generic_workspace_dir self.workspace_dir = generic_workspace_dir(self.user_id) try: os.makedirs(self.workspace_dir, exist_ok=True) except Exception: pass # 8. 五级作用域工具解析（2026-09-10）：global(GENERAL_TOOLS) → org(策略+机构技能包) # → pipeline(能力包+产线策略) → role(RoleSpec 白名单+角色策略) → project(策略微调)。 # 产出：registry 重建为解析后工具集 + _allowed_tools 执行层门禁 # （LLM 幻觉调用作用域外工具名 → 直接拒绝，schema 防线挡不住幻觉）。 try: await self._resolve_tool_scopes() except Exception as e: # 解析失败不阻断会话：回退旧行为（registry 保持 config.tools 全量，无门禁） logger.warning(f"tool scope resolve failed, fallback to base tools: {e}") self._allowed_tools = None async def _resolve_tool_scopes(self): """五级作用域工具解析（pipeline_core.tool_sources）。 - merged_skills 用与技能注入完全相同的六级可见性参数（generic 只 global）， 技能作用域即工具作用域：机构装的技能包声明的 capability 工具只对该机构可见。 - capability 工具注册进本会话 registry（合成 ToolDefinition + 路由 handler → capability_tools.exec_capability_tool），native FC schema 自动带上。 - 策略表 pipeline_tool_policies 缺表/查询失败 → 按无策略继续（load_policies 内降级）。 """ from pipeline_core.tool_sources import resolve_scoped_tools # 与 _build_system_prompt 技能注入同款的可见性参数 pipeline_id = "" if self.generic else self.pipeline_id role = "" if self.generic else (self.role or "") project_id = "" if self.generic else self.project_id org_id = "" if self.generic else (self.org_id or "") user_id = "" if self.generic else (self.user_id or "") merged = None if self.config.skills.enabled and self._skill_loader: try: merged = self._skill_loader.get_merged( pipeline_id=pipeline_id, role=role, project_id=project_id, org_id=org_id, user_id=user_id) except Exception as e: logger.warning(f"get_merged for tool scopes failed: {e}") from sqlor.dbpools import DBPools db = DBPools() async with db.sqlorContext("pipeline") as sor: res = await resolve_scoped_tools( sor, base_tools=self.config.tools, merged_skills=merged, generic=self.generic, org_id=org_id, user_id=user_id, pipeline_id=pipeline_id, role=role, project_id=project_id) self._tool_scope_trace = res.trace self._capability_tool_names = set(res.capability_tools or []) self._allowed_tools = set(res.allowed) # registry 重建：解析后的 ToolDefinition 集 + capability 工具合成注册 if self._tool_registry is not None: from pipeline_core.tool_registry import ToolRegistry from pipeline_core.agent_config import ToolDefinition reg = ToolRegistry() for t in res.tools: reg.register(t) # capability 工具：schema 从 capability_tools.TOOL_SCHEMAS 取， # handler 路由到 exec_capability_tool（系统上下文自动注入，不暴露给 LLM） for name in sorted(self._capability_tool_names): if reg.get(name) is not None: continue try: from .capability_tools import TOOL_SCHEMAS schema = TOOL_SCHEMAS.get(name) if not schema: continue # 声明了但无实现 schema：跳过（不注册幻觉入口） td = ToolDefinition( name=name, description=schema["description"], parameters=dict(schema.get("params") or {}), category="capability", required=list(schema.get("required") or []), ) reg.register(td) except Exception as e: logger.warning(f"capability tool {name} register failed: {e}") self._tool_registry = reg logger.info("tool scopes resolved: allowed=%d (capability=%d) trace=%s", len(self._allowed_tools), len(self._capability_tool_names), " | ".join(res.trace)) async def _resolve_skills_base_dir(self): """技能根目录 = 全局 skills/（单一技能树，所有机构共享读，2026-08-21 重构）。""" try: from pipeline_core.skill_pack import get_skills_base return get_skills_base() except Exception: return self.config.skills.base_dir async def _llm_select_skills(self, user_input: str, catalog, max_skills: int) -> List[str]: """用 LLM 从技能目录选最相关的 N 个技能（语义匹配，非关键词硬编码）。""" if not catalog: return [] catalog_text = "\n".join(f"- {name}: {desc}" for name, desc in catalog) prompt = ( f"以下是可用技能目录（技能名: 描述）：\n{catalog_text}\n\n" f"用户需求：{user_input}\n\n" f"请从上述目录中选出与用户需求最相关的至多 {max_skills} 个技能，" f"只返回技能名的 JSON 数组，如 [\"a\", \"b\"]。没有相关技能返回 []。" ) try: from pipeline_service.llm_bridge import llm_call_msgs import json as _json import re as _re content = await llm_call_msgs( [{"role": "user", "content": prompt}], model=self.model_name, temperature=0, org_id=self.org_id, purpose='utility', session_id=self.session_id, timeout=self._UTILITY_TIMEOUT, ) m = _re.search(r"\[[^\]]*\]", content or "") if m: names = _json.loads(m.group(0)) return [n for n in names if isinstance(n, str)] return [] except Exception as e: logger.error(f"_llm_select_skills failed: {e}") return [] async def _build_system_prompt(self, user_input: str) -> str: """组装完整 system prompt = 基础 prompt + 记忆 + 技能 + 工具列表""" prompt = self.config.system_prompt # 注入当前项目上下文 proj_name = self.project_id try: ctx = await self._load_project_context() if ctx: proj_name = ctx.replace("项目: ", "") except Exception: pass prompt += f"\n当前项目: {proj_name}\n" if proj_name else "" # 注入记忆（分域：通用 + 产线 + 项目 叠加） # 产线隔离（2026-09-05）：纯通用会话只读 global 通用记忆，绝不带 # 「产线+项目」叠加——否则通用助手会看到产线专属记忆。 if self.config.memory.enabled and self._memory_store: # 多租户可见性版（2026-09-10）：传 org_id/user_id → store 按 # visible_to 过滤（平台种子 + 本机构 org/pipeline/project + 本人 user）， # 替代旧的「global + pipeline 叠加」——旧版会把其他机构的 # org/pipeline/project 归属记忆也注入本会话（跨机构泄漏）。 if self.generic: # 通用会话：只注入种子 + 本人 user 域 + 本机构 org 域 mem_block = await self._memory_store.build_prompt_block( max_entries=15, org_id=self.org_id, user_id=self.user_id) else: mem_block = await self._memory_store.build_prompt_block( max_entries=20, org_id=self.org_id, user_id=self.user_id, pipeline_id=self.pipeline_id, project_id=self.project_id) if mem_block: prompt += f"\n\n## 持久记忆\n{mem_block}" # 注入技能（六级隔离：global→org→pipeline→role→project→user） if self.config.skills.enabled and self._skill_loader: # 产线隔离（2026-09-05）：纯通用会话只检索 global 技能， # 不挂产线/角色/项目/机构技能。 pipeline_id = "" if self.generic else (self.pipeline_id if self.config.skills.enable_pipeline else "") role = "" if self.generic else (self.role if self.config.skills.enable_role else "") project_id = "" if self.generic else (self.project_id if self.config.skills.enable_project else "") org_id = "" if self.generic else (self.org_id if self.config.skills.enable_org else "") user_id = "" if self.generic else (self.user_id if self.config.skills.enable_user else "") # 技能检索必须 LLM 做（语义匹配，非关键词硬编码） catalog = self._skill_loader.get_skill_catalog( pipeline_id=pipeline_id, role=role, project_id=project_id, org_id=org_id, user_id=user_id) names = await self._llm_select_skills( user_input, catalog, self.config.skills.max_skills_per_turn) if names: skill_block = self._skill_loader.build_prompt_block_by_names( names, pipeline_id=pipeline_id, role=role, project_id=project_id, org_id=org_id, user_id=user_id) else: # LLM 无结果兜底：按 scope+名字列出前 N 个（不传 user_input，走 fallback 分支） skill_block = self._skill_loader.build_prompt_block( pipeline_id=pipeline_id, role=role, project_id=project_id, org_id=org_id, user_id=user_id, max_skills=self.config.skills.max_skills_per_turn) if skill_block: prompt += f"\n\n{skill_block}" # 注入工具列表（先构建，再替换占位符） tools_text = "" if self._tool_registry: tools_text = self._tool_registry.to_text_description() else: lines = [] for t in self.config.tools: if t.enabled: params = ", ".join(f"{k}: {v}" for k, v in (t.parameters or {}).items()) lines.append(f"- {t.name}({params}): {t.description}") tools_text = "\n".join(lines) prompt = prompt.replace("{project_name}", proj_name) prompt = prompt.replace("{tools_description}", tools_text) # 敏感信息使用铁律（2026-09-17）：仅在有真实用户身份时注入—— # 无人值守/内部 agent 不挂 secret 工具，也就不给凭据使用引导。 if self.user_id: try: from .secret_vault import SECRET_PROMPT_BLOCK prompt += SECRET_PROMPT_BLOCK except Exception as e: logger.warning(f"inject secret prompt block failed: {e}") return prompt # ═══════════════════════════════════════════════════════ # LLM 调用 # ═══════════════════════════════════════════════════════ async def _call_llm(self) -> dict: """调用 LLM，返回 {"content": str, "tool_calls": [...]}。 优先使用 OpenAI native function calling。若 registry 不可用或调用失败， 回退到纯文本调用（content 为原始文本，tool_calls 为空）。 """ from pipeline_service.llm_bridge import llm_call_msgs_native tools_schema = None if self._tool_registry: try: tools_schema = self._tool_registry.to_openai_schema() except Exception as e: logger.error(f"to_openai_schema failed: {e}") if tools_schema: try: return await llm_call_msgs_native( self._msgs, tools=tools_schema, model=self.model_name, temperature=self.config.temperature, org_id=self.org_id, session_id=self.session_id, timeout=self._MAIN_TIMEOUT, ) except Exception as e: # 原生视觉降级（2026-09-10）：模型不支持多模态 content 数组时 # 上游会报错——剥离图片 parts 重试一次，并注入说明让 LLM 如实 # 告知用户「当前模型看不了图」（绝不静默丢图假装看过了）。 degraded = self._degrade_images_if_needed(e) if degraded: try: return await llm_call_msgs_native( self._msgs, tools=tools_schema, model=self.model_name, temperature=self.config.temperature, org_id=self.org_id, session_id=self.session_id) except Exception: pass logger.error(f"native function calling failed, fallback to text: {e}") # 回退：纯文本调用（content 为原始文本） from pipeline_service.llm_bridge import llm_call_msgs try: content = await llm_call_msgs( self._msgs, model=self.model_name, temperature=self.config.temperature, org_id=self.org_id, session_id=self.session_id, timeout=self._MAIN_TIMEOUT, ) except Exception as e: if self._degrade_images_if_needed(e): content = await llm_call_msgs( self._msgs, model=self.model_name, temperature=self.config.temperature, org_id=self.org_id, session_id=self.session_id, timeout=self._MAIN_TIMEOUT) else: raise return {"content": content or "", "tool_calls": []} def _degrade_images_if_needed(self, err: Exception) -> bool: """含图消息调用失败时剥离图片 parts 重试（一次性降级）。 只对「消息里确实有图」的失败生效；剥离后就地改写 self._msgs （多模态 content 数组 → 纯文本 + 说明），返回是否执行了降级。 """ if not self._has_images: return False self._has_images = False # 只降级一次 n_imgs = 0 for m in self._msgs: c = m.get("content") if isinstance(c, list): texts = [pt.get("text", "") for pt in c if pt.get("type") == "text"] n_imgs += sum(1 for pt in c if pt.get("type") == "image_url") m["content"] = ("\n".join(t for t in texts if t) + f"\n\n[系统说明：用户本次上传了 {n_imgs} 张图片，但当前模型" f"（{self.model_name or '默认模型'}）调用图片失败" f"（{str(err)[:120]}），图片已剥离。请如实告知用户当前模型" f"无法查看图片，可建议切换支持视觉的模型，或改用 " f"invoke_model 的 i2t 图像理解能力。不要假装看过图片。]") logger.warning(f"image degrade: stripped {n_imgs} images after LLM failure: {err}") return n_imgs > 0 # ═══════════════════════════════════════════════════════ # 解析 LLM 输出 # ═══════════════════════════════════════════════════════ def _parse_actions(self, raw: str) -> List[dict]: """解析 LLM 输出，支持单个 JSON 或连续多个 JSON。 对照 HA：解析器成熟处理多 JSON + 容错。 """ raw = (raw or "").strip() # 去除 markdown 代码块包裹 if raw.startswith("```"): parts = raw.split("\n", 1) if len(parts) > 1: raw = parts[1] if raw.endswith("```"): raw = raw[:-3] raw = raw.strip() results = [] # 尝试多 JSON 解析 # LLM 可能输出: {"action":"tool_call",...}\n{"action":"tool_call",...} lines = raw.split("\n") buffer = "" depth = 0 for line in lines: line_stripped = line.strip() if not line_stripped: if buffer: # 尝试解析 buffer result = self._try_parse_json(buffer) if result: results.append(result) buffer = "" continue buffer += line_stripped # 简单大括号计数 depth += line_stripped.count("{") - line_stripped.count("}") if depth == 0 and buffer: result = self._try_parse_json(buffer) if result: results.append(result) buffer = "" # 处理残留 if buffer: result = self._try_parse_json(buffer) if result: results.append(result) if not results: # 完全无法解析 JSON → 当作纯文本回复 results.append({"action": "reply", "message": raw}) return results def _try_parse_json(self, text: str) -> Optional[dict]: """容错 JSON 解析""" text = text.strip() try: d = json.loads(text) if isinstance(d, dict): return d except (json.JSONDecodeError, ValueError): # 尝试提取第一个 JSON 对象 for start_char in ["{", "["]: if start_char in text: idx = text.index(start_char) end_char = "}" if start_char == "{" else "]" end_idx = text.rfind(end_char) if end_idx > idx: try: d = json.loads(text[idx:end_idx + 1]) if isinstance(d, dict): return d except (json.JSONDecodeError, ValueError): pass return None # ═══════════════════════════════════════════════════════ # 工具执行 # ═══════════════════════════════════════════════════════ async def _execute_tool(self, tool_name: str, params: dict) -> str: """执行工具，返回结果字符串。 优先级： 1. ToolRegistry 注册的 handler 2. SDLC 内建工具（兼容 cockpit_chat.dspy 的 TOOLS） 3. ask_user（始终可用） """ # 别名归一化：LLM 会编造工具名（2026-08-31 项目管理的常见变体）， # 映射到真实工具，避免「未知工具」导致用户指令落空。 tool_name = _PROJECT_TOOL_ALIASES.get(tool_name, tool_name) # 0. 五级作用域执行门禁（2026-09-10）：schema 里不给看挡不住幻觉—— # LLM 仍可能喊出作用域外工具名（generic 喊 create_task、A 产线喊 B 产线 # 工具、他机构技能包的工具）。必须在分发前按解析出的允许集硬拦， # 拒绝并回可行动提示（列出本会话真实可用工具），而不是「未知工具」让 LLM 瞎猜。 # _allowed_tools 为 None 时（解析失败降级）不拦截，回退旧行为。 if self._allowed_tools is not None and tool_name not in self._allowed_tools: avail = sorted(self._allowed_tools) shown = ", ".join(avail[:20]) + ("..." if len(avail) > 20 else "") scope_desc = "通用会话" if self.generic else ( f"产线 {self.pipeline_id}" + (f"/角色 {self.role}" if self.role else "")) return (f"FAIL: 工具 {tool_name} 不在本会话（{scope_desc}）的可用范围内，已拒绝执行。" f"本会话可用工具: {shown}") # 1. 注册的 handler if self._tool_registry: handler = self._tool_registry.get_handler(tool_name) if handler: try: result = await handler(params, { "project_id": self.project_id, "user_id": self.user_id, "workspace_dir": self.workspace_dir, }) return str(result) except Exception as e: logger.error(f"Tool {tool_name} handler error: {e}") return f"ERROR: {str(e)[:300]}" # 1.5 capability 工具（技能 frontmatter 驱动，2026-09-10）： # 路由到 capability_tools.exec_capability_tool（按 TOOL_SCHEMAS 分发到 # 对应 capability 模块，系统上下文 project_id/iteration_id/who/agent_id # 自动注入，不暴露给 LLM）。generic 会话不挂（作用域解析已排除）。 # ⚠️ 必须「名字被声明 且 TOOL_SCHEMAS 有 schema」双守卫（对齐 v1 # _exec_agent_tool 的 `if tool in TOOL_SCHEMAS`）：技能 frontmatter 声明的 # capability 工具（task 技能的 claim_task/reset_task_retry 等）多数不在 # TOOL_SCHEMAS（那是概念工具 schema 表），真正实现是产线能力包 handler # （第 2 步 _execute_ability_tool）。此前只查名字 → 整族任务状态机工具被 # 本步劫持，返回「FAIL: 未注册的能力工具」，能力包 handler 永远到不了 # （2026-09-14 pbls 项目实测：主 agent 反复调 reset_task_retry 全 FAIL）。 if tool_name in self._capability_tool_names: from .capability_tools import TOOL_SCHEMAS, exec_capability_tool if tool_name in TOOL_SCHEMAS: return await exec_capability_tool(tool_name, params, self._build_ctx()) # 声明了但无概念工具 schema → 落到第 2 步产线能力包分发（真实 handler 所在） # 2. 产线能力包工具（从 PipelineAbility 注册表按 pipeline_id 取 handler） result = await self._execute_ability_tool(tool_name, params) if result is not None: return result # 3. 通用内建工具（core 内核：项目管理/终端/文件/搜索/会话/规划） result = await self._execute_sdlc_tool(tool_name, params) if result is not None: return result # 4. ask_user if tool_name == "ask_user": return f"QUESTION: {params.get('question', '')}" return f"未知工具: {tool_name}" def _build_ctx(self) -> dict: """构造传给产线能力 handler / slash 命令的上下文。""" return { "project_id": self.project_id, "user_id": self.user_id, "org_id": self.org_id, "pipeline_id": self.pipeline_id, "space": self.space, "workspace_dir": self.workspace_dir, "model_name": self.model_name, "config": self.config, # 会话级隔离键（2026-09-05）：能力 handler 的会话态缓存（如 extract 规格锚定） # 必须按 session_id 隔离，防跨会话串数据 "session_id": self.session_id, } def _build_slash_ctx(self) -> dict: """构造传给 slash 命令 handler 的上下文（含 executor 引用 + role + base_url）。""" ctx = self._build_ctx() ctx["executor"] = self ctx["role"] = self.role ctx["base_url"] = getattr(self, "base_url", "") or "" return ctx async def _execute_ability_tool(self, tool_name: str, params: dict) -> Optional[str]: """产线能力包工具：按 pipeline_id 从 PipelineAbility 注册表取 handler 执行。""" # 产线隔离（2026-09-05）：纯通用会话不挂任何产线能力包，执行层直接拒绝。 if self.generic: return None try: from pipeline_core import get_ability, DEFAULT_ABILITY_ID ability = get_ability(self.pipeline_id) or get_ability(DEFAULT_ABILITY_ID) if not ability or tool_name not in ability.handlers: return None from sqlor.dbpools import DBPools db = DBPools() async with db.sqlorContext("pipeline") as sor: return await ability.handlers[tool_name](sor, params, self._build_ctx()) except Exception as e: logger.error(f"ability tool {tool_name} error: {e}") return f"ERROR: {str(e)[:300]}" async def _execute_sdlc_tool(self, tool_name: str, params: dict) -> Optional[str]: """通用内建工具（core 内核，产线无关）。""" # 导入现有 handler try: from sqlor.dbpools import DBPools db = DBPools() async with db.sqlorContext("pipeline") as sor: return await self._dispatch_sdlc_tool(sor, tool_name, params) except Exception as e: logger.error(f"SDLC tool {tool_name} error: {e}") return f"ERROR: {str(e)[:300]}" async def _dispatch_sdlc_tool(self, sor, tool_name: str, params: dict) -> str: """SDL 工具分发（兼容 cockpit_chat.dspy 的 16 个工具）""" p = params or {} pid = self.project_id # 产线隔离（2026-09-05）：纯通用会话拒绝一切项目管理工具——此时 # tool_name 已经过 _PROJECT_TOOL_ALIASES 别名归一，别名变体同样被拦。 # schema 层（load_agent_config 剔除）+ 执行层（这里）双保险。 if self.generic and tool_name in _PROJECT_TOOL_NAMES: return ("FAIL: 当前是通用助手会话（未挂产线插件），不提供项目管理能力，" "无法查看或操作任何产线的项目。") # 产线耦合工具拦截（2026-09-05 第 6 层）：run_command 的 shell 无法圈禁 # 在工作目录内，通用会话一律拒绝。 if self.generic and tool_name in _GENERIC_DENIED_TOOLS: return ("FAIL: 当前是通用助手会话，不提供命令执行能力。" "如需处理文件，可直接上传文件，我会读取上传内容作答。") # 通用内建工具映射（core 内核，产线无关）。 # 产线专属工具（create_task/diagnose_project 等）已迁移到 sdlc_ability 能力包， # 由 _execute_ability_tool 按 pipeline_id 动态挂载。 handlers = { "switch_project": self._t_switch_project, "create_project": self._t_create_project, # ── 项目管理（终止/删除类代码层校验人类 owner）── "project_info": self._t_project_info, "list_my_projects": self._t_list_my_projects, "pause_project": self._t_pause_project, "resume_project": self._t_resume_project, "delete_project": self._t_delete_project, "run_command": self._t_run_command, "process": self._t_process, # ── 通用工具集（Hermes CLI 能力子集）── "read_file": self._t_read_file, "patch_file": self._t_patch_file, "load_skill": self._t_load_skill, "list_packs": self._t_list_packs, "install_pack": self._t_install_pack, "propose_skill": self._t_propose_skill, "write_file": self._t_write_file, "list_files": self._t_list_files, "search_files": self._t_search_files, "session_search": self._t_session_search, "memory": self._t_memory, "manage_skill": self._t_manage_skill, "todo": self._t_todo, "delegate_subtask": self._t_delegate_subtask, "subagent": self._t_subagent, # ── 平台模型（2026-09-07：按任务自动选型 + 全能力调用）── "list_platform_models": self._t_list_platform_models, "invoke_model": self._t_invoke_model, # ── 联网检索/网页抓取（2026-09-08，甲类只读能力，SSRF 防护在 web_tools）── "web_search": self._t_web_search, "fetch_url": self._t_fetch_url, # ── 项目数据只读查询（2026-09-08，白名单+强制项目过滤在 db_query）── "query_project_data": self._t_query_project_data, # ── 用户敏感信息库（2026-09-17）：签名适配 (sor,params,ctx) ← (sor,p,pid) ── # ctx 用 _build_ctx()（含真实 self.user_id/self.org_id）——secret 归属必须 # 是真实身份，不能用作用域解析里被 generic 置空的那份。 **{n: (lambda _n: (lambda s, _p, _pid: self._t_secret_tool(s, _n, _p)))(n) for n in _SECRET_TOOL_NAMES}, } handler = handlers.get(tool_name) if handler: return await handler(sor, p, pid) return None async def _t_secret_tool(self, sor, tool_name: str, params: dict) -> str: """敏感信息工具统一入口（2026-09-17）。 身份强制来自会话真实上下文（self.user_id / self.org_id），忽略 LLM 传的任何 归属参数——防越权读写他人 secret。无 user_id（无人值守/内部 agent）一律拒绝。 """ from . import secret_vault if not self.user_id: return "FAIL: 当前会话没有用户身份，不提供敏感信息能力（防无人值守场景泄露凭据）" h = secret_vault.SECRET_HANDLERS.get(tool_name) if not h: return f"FAIL: 未知敏感信息工具 {tool_name}" try: _r = await h(sor, params or {}, {"user_id": self.user_id, "org_id": self.org_id or "", "project_id": self.project_id}) # 写操作必须显式 COMMIT（多 worker 架构：不提交则其他进程读不到刚存的凭据）。 # 与本文件其他写库工具（_persist_project 等 6 处）同款做法；只读工具多跑一次 # COMMIT 无害（sqlor 事务幂等），省去按工具名区分读写。 try: await sor.sqlExe("COMMIT", {}) except Exception as _ce: logger.warning(f"secret tool {tool_name} commit failed: {_ce}") return _r except Exception as e: logger.error(f"secret tool {tool_name} error: {e}") return f"ERROR: {str(e)[:300]}" # ── 工具实现 ── async def _persist_project(self, sor, pid): """持久化当前项目。 - 有 session_id：写 pipeline_session_settings（web 多 tab 各自项目上下文，互不覆盖）， 同时写全局 pipeline_agent_settings.current_project_id 作为「最近项目」兜底（供无 session 消费者）。 - 无 session_id：只写全局（向后兼容，如微信通道 / run_agent 直连）。 只改 self.project_id 不够——AgentExecutor 每轮新建，run 结束即销毁， 下一轮又从持久层读回旧项目。 """ if not self.user_id or not pid: return try: from appPublic.uniqueID import getID # 全局「最近项目」兜底 await sor.sqlExe( "UPDATE pipeline_agent_settings SET current_project_id=${pid}$ " "WHERE user_id=${uid}$", {"pid": pid, "uid": self.user_id}) exists = await sor.sqlExe( "SELECT 1 FROM pipeline_agent_settings WHERE user_id=${uid}$", {"uid": self.user_id}) if not exists: await sor.sqlExe( "INSERT INTO pipeline_agent_settings (id, user_id, current_project_id) " "VALUES (${id}$, ${uid}$, ${pid}$)", {"id": getID(), "uid": self.user_id, "pid": pid}) # 会话级项目上下文（多 tab 隔离） if self.session_id: try: await sor.sqlExe( "UPDATE pipeline_session_settings SET current_project_id=${pid}$ " "WHERE user_id=${uid}$ AND session_id=${sid}$", {"pid": pid, "uid": self.user_id, "sid": self.session_id}) sexists = await sor.sqlExe( "SELECT 1 FROM pipeline_session_settings " "WHERE user_id=${uid}$ AND session_id=${sid}$", {"uid": self.user_id, "sid": self.session_id}) if not sexists: await sor.sqlExe( "INSERT INTO pipeline_session_settings " "(id, user_id, session_id, current_project_id) " "VALUES (${id}$, ${uid}$, ${sid}$, ${pid}$)", {"id": getID(), "uid": self.user_id, "sid": self.session_id, "pid": pid}) except Exception as e: logger.warning(f"persist session project failed: {e}") except Exception as e: logger.warning(f"persist project failed: {e}") async def _t_switch_project(self, sor, p, pid): name = p.get("project_name", "").strip() if not name: return "需要项目名称" # 精确匹配（限定当前项目空间 + 机构） recs = await sor.sqlExe( "SELECT id, name FROM sd_projects WHERE name=${n}$ AND pipeline_id=${s}$", {"n": name, "s": self.space}) if recs: r = recs[0] self.project_id = getattr(r, "id", "") await self._persist_project(sor, self.project_id) await self._refresh_workspace_dir(sor) return f"OK: 已切换到 {getattr(r, 'name', name)}" # LLM 分类匹配 try: from pipeline_service.llm_bridge import llm_call all_recs = await sor.sqlExe( "SELECT name FROM sd_projects WHERE pipeline_id=${s}$ ORDER BY created_at DESC LIMIT 20", {"s": self.space}) pnames = [getattr(r, "name", "") for r in (all_recs or [])] classify_prompt = f"用户输入: {name}\n项目列表: {', '.join(pnames)}\n\n判断用户想要哪个项目。只回复项目名或\"不存在\"。" matched = await llm_call(classify_prompt, temperature=0.0, org_id=self.org_id, purpose='utility', session_id=self.session_id) matched = matched.strip().strip('"').strip("'") recs2 = await sor.sqlExe( "SELECT id, name FROM sd_projects WHERE name=${n}$ AND pipeline_id=${s}$", {"n": matched, "s": self.space}) if recs2: r = recs2[0] self.project_id = getattr(r, "id", "") await self._persist_project(sor, self.project_id) await self._refresh_workspace_dir(sor) return f"OK: 已切换到 {getattr(r, 'name', matched)}" except Exception: pass return f"未找到项目: {name}" async def _refresh_workspace_dir(self, sor): """切换项目后同步刷新工作目录（与上传落盘/初始化同一解析逻辑）。 不刷新则 list_files/read_file 仍停在切换前的目录（或无项目时的总根）。 """ try: from .workspace import get_project_dir_by_id _pdir, _ = await get_project_dir_by_id(sor, self.project_id) if _pdir: import os as _os _os.makedirs(_pdir, exist_ok=True) self.workspace_dir = _pdir except Exception: pass async def _t_create_project(self, sor, p, pid): import os import re from appPublic.uniqueID import getID name = p.get("name", "").strip() desc = p.get("description", "") if not name: return "需要项目名称" # 查当前用户 org_id（工作空间按 org 隔离） org_id = "0" if self.user_id: _u = await sor.sqlExe( "SELECT orgid FROM users WHERE id=${u}$ LIMIT 1", {"u": self.user_id}) if _u: org_id = getattr(_u[0], "orgid", "0") or "0" # 项目专属工作空间目录（统一机制：{base}/{org}/{space}/projects/{项目名}/， # 用名字不用 id；创建时定死并写库，一切解析只读库不推导）。 from .workspace import alloc_project_dir pid_val = getID() project_dir, slug = await alloc_project_dir(sor, org_id, self.space, name, pid_val) await sor.C("sd_projects", { "id": pid_val, "name": name, "description": desc, "status": "active", "org_id": org_id, "pipeline_id": self.space, "workspace_dir": project_dir, "directory_name": slug, "created_by": self.user_id or "", "default_model": self.model_name or "", }) await sor.C("sd_iterations", { "id": getID(), "project_id": pid_val, "iteration_name": f"{name}-初始迭代", "iteration_type": "default", "status": "in_progress", "priority": 1, "seq_no": 1, }) self.project_id = pid_val await self._persist_project(sor, pid_val) return f"OK: 已创建项目 {name}" # ── 项目管理工具（2026-08-31 新增：会话 agent 获得项目管理能力）── # 安全模型：终止（暂停）/删除等指令必须来自项目的人类 owner。 # 校验在代码层做（_resolve_owned_project 内调 check_project_owner）， # 不依赖 prompt——LLM 可以被注入/诱导，但校验函数只认数据库事实。 # owner 定义：sd_projects.created_by == 当前登录用户（self.user_id）。 async def _resolve_owned_project(self, sor, p): """解析目标项目并校验操作者是否为人类 owner。 project_name 缺省时用当前项目（self.project_id）；给了名称则在当前产线 空间内精确匹配。返回 (project_id, project_name, owner_err)： owner_err 非空 = 无权限或项目不存在，调用方必须原样拒绝。 """ from .project_capability import check_project_owner name = (p.get("project_name") or "").strip() if not name: pid = self.project_id if not pid: return "", "", "当前会话没有选中项目，请先说明要操作哪个项目（或先切换项目）" recs = await sor.sqlExe( "SELECT id, name, created_by FROM sd_projects WHERE id=${pid}$", {"pid": pid}) await sor.sqlExe("COMMIT", {}) if not recs: return "", "", "当前项目已不存在" pname = getattr(recs[0], "name", "") or pid else: recs = await sor.sqlExe( "SELECT id, name, created_by FROM sd_projects " "WHERE name=${n}$ AND pipeline_id=${s}$", {"n": name, "s": self.space}) await sor.sqlExe("COMMIT", {}) if not recs: return "", "", f"项目不存在：{name}（当前产线空间 {self.space}）" pid = getattr(recs[0], "id", "") pname = getattr(recs[0], "name", "") or pid # 代码层 owner 校验：指令必须来自该项目的人类 owner（创建者） ok, err = await check_project_owner(pid, self.user_id or "", sor) if not ok: return "", "", f"权限不足：{err}（项目「{pname}」的 owner 才可执行此操作）" return pid, pname, "" async def _t_project_info(self, sor, p, pid): tpid, pname, err = await self._resolve_owned_project(sor, p) if err: return f"FAIL: {err}" recs = await sor.sqlExe( "SELECT id, name, description, status, pipeline_id, org_id, " "created_by, created_at, updated_at FROM sd_projects WHERE id=${pid}$", {"pid": tpid}) await sor.sqlExe("COMMIT", {}) if not recs: return "FAIL: 项目不存在" r = recs[0] lines = [ f"项目：{getattr(r, 'name', '')}", f"状态：{getattr(r, 'status', '')}", f"描述：{getattr(r, 'description', '') or '（无）'}", f"产线：{getattr(r, 'pipeline_id', '') or '（未绑定）'}", f"创建：{getattr(r, 'created_at', '')} ｜ 更新：{getattr(r, 'updated_at', '')}", ] # 迭代概况 iters = await sor.sqlExe( "SELECT iteration_name, status FROM sd_iterations WHERE project_id=${pid}$ ORDER BY seq_no", {"pid": tpid}) await sor.sqlExe("COMMIT", {}) if iters: lines.append("迭代：" + "；".join( f"{getattr(i, 'iteration_name', '')}({getattr(i, 'status', '')})" for i in iters[:10])) return "\n".join(lines) async def _t_list_my_projects(self, sor, p, pid): # 只列当前用户创建的项目（不暴露他人项目） uid = self.user_id or "" if not uid: return "FAIL: 未登录，无法列出你的项目" status = (p.get("status") or "").strip() sql = ("SELECT id, name, status, pipeline_id, created_at FROM sd_projects " "WHERE created_by=${uid}$") params = {"uid": uid} if status: sql += " AND status=${st}$" params["st"] = status sql += " ORDER BY created_at DESC LIMIT 30" recs = await sor.sqlExe(sql, params) await sor.sqlExe("COMMIT", {}) if not recs: return f"你名下没有项目{('（状态=' + status + '）') if status else ''}" lines = [] for r in recs: lines.append(f"- {getattr(r, 'name', '')} ｜ {getattr(r, 'status', '')} ｜ " f"产线 {getattr(r, 'pipeline_id', '') or '—'} ｜ {getattr(r, 'created_at', '')}") return f"你创建的项目（{len(lines)} 个）：\n" + "\n".join(lines) async def _t_pause_project(self, sor, p, pid): tpid, pname, err = await self._resolve_owned_project(sor, p) if err: return f"FAIL: {err}" from .project_capability import pause_project ok, msg = await pause_project(tpid, who=self.user_id, agent_id="cockpit") if not ok: return f"FAIL: {msg}" logger.info("cockpit pause_project: %s by user=%s", tpid, self.user_id) return f"OK: 项目「{pname}」已暂停（终止推进）。需要时可用 resume_project 恢复。" async def _t_resume_project(self, sor, p, pid): tpid, pname, err = await self._resolve_owned_project(sor, p) if err: return f"FAIL: {err}" from .project_capability import resume_project ok, msg = await resume_project(tpid, who=self.user_id, agent_id="cockpit") if not ok: return f"FAIL: {msg}" logger.info("cockpit resume_project: %s by user=%s", tpid, self.user_id) return f"OK: 项目「{pname}」已恢复推进。" async def _t_delete_project(self, sor, p, pid): confirm = p.get("confirm") if str(confirm).lower() not in ("true", "1", "yes"): return ("FAIL: 参数错误，请直接带 confirm=true 调用（平台确认门会弹窗向用户最终确认，" "无需你先向用户复述确认）。") tpid, pname, err = await self._resolve_owned_project(sor, p) if err: return f"FAIL: {err}" # 删除后清掉本会话的项目上下文（否则留下悬空指针，下次打开报「请先切换项目」） from .project_capability import delete_project ok, msg = await delete_project(tpid, who=self.user_id, agent_id="cockpit", confirm=True) if not ok: return f"FAIL: {msg}" if self.project_id == tpid: self.project_id = "" # 清会话级+全局「当前项目」指针（_persist_project 对空 pid 直接 return， # 这里直接清，否则留下悬空指针，下次打开报「请先切换项目」） try: await sor.sqlExe( "UPDATE pipeline_session_settings SET current_project_id='' " "WHERE user_id=${uid}$ AND session_id=${sid}$", {"uid": self.user_id or "", "sid": self.session_id or ""}) await sor.sqlExe( "UPDATE pipeline_agent_settings SET current_project_id='' " "WHERE user_id=${uid}$", {"uid": self.user_id or ""}) await sor.sqlExe("COMMIT", {}) except Exception as e: logger.warning(f"clear project pointer after delete failed: {e}") logger.info("cockpit delete_project: %s by user=%s", tpid, self.user_id) return f"OK: 项目「{pname}」已删除（含归档备份）。{msg}" async def _t_run_command(self, sor, p, pid): cmd = p.get("command", "") if not cmd: return "需要命令" try: from pipeline_service.agent_loop import _run_shell, _find_bwrap # 通用会话（generic）strict 沙箱档（2026-09-08 乙类开放）： # bwrap 圈死——平台目录（/d/pipeline、/d/doit）完全不挂载（不可见）， # 可写根 = 用户自己的 _general/{uid} 目录。旧禁令的理由「shell 无法 # 圈禁在工作目录内、可枚举全工作空间」已被 strict 档根治（连读都不可见）。 # 无 bwrap 时 generic 一律拒绝（绝不降级为裸 shell——安全优先）。 if self.generic and not _find_bwrap(): return ("FAIL: 通用会话的命令执行需要 bwrap 沙箱（当前服务器不可用），已拒绝执行。" "文件类操作请改用 read_file/write_file/list_files/search_files。") # 敏感信息执行边界（2026-09-17）：占位符 @@sec:NAME@@ → $PIPELINE_SEC_NAME， # 真值经 env 注入子进程，**不进命令字符串**（命令可安全落库/进上下文）。 # 未知占位符原样保留（绝不替换成空串——空串会静默改变命令语义）。 # 必须在前台/后台分支**之前**处理，两条执行路径才都受保护。 secret_env = None try: from . import secret_vault cmd, secret_env, unknown = await secret_vault.prepare_command( sor, cmd, org_id=self.org_id or "", user_id=self.user_id or "") if unknown: return ("FAIL: 命令引用了不存在的敏感信息 " + ", ".join(unknown) + "（已拒绝执行，防止空值改变命令语义）。用 list_secrets 核对名称。") except Exception as e: logger.warning(f"secret prepare_command failed (run without secrets): {e}") # PII 执行边界（2026-09-17）：@@pii:占位符 → $PIPELINE_PII_*，真值经 env 注入， # 与凭据同构；缓存缺失时引用保留（命令取空，不静默改语义——与凭据不同：PII # 缺失是过期常态，拒绝执行会误伤，取空+输出可见更安全）。 pii_env = None if self.pii_session_key and "@@pii:" in cmd: try: from . import pii_guard cmd, pii_env = await pii_guard.prepare_command( cmd, self.pii_session_key) except Exception as e: logger.warning(f"pii prepare_command failed (run without pii env): {e}") if pii_env: secret_env = {**(secret_env or {}), **pii_env} # 后台执行（2026-09-10，对齐 Hermes terminal background）： # 状态文件化到 workspace/.bg/，跨 worker 进程可 poll； # 沙箱档位与前台完全一致（generic 强制 strict，无 bwrap 上面已拒）。 bg = str(p.get("background", "")).strip().lower() if bg in ("true", "1", "yes"): from . import bg_jobs try: job_id = await bg_jobs.start_bg_job( cmd, self.workspace_dir, strict=self.generic, secret_env=secret_env) except bg_jobs.BgJobError as e: return f"FAIL: {e}" return (f"OK: 后台任务已启动 job_id={job_id}。" f"用 process(action=poll|log|wait|kill, job_id={job_id}) 跟进输出与状态。") try: timeout = int(p.get("timeout") or 60) except (TypeError, ValueError): timeout = 60 timeout = max(5, min(timeout, 300)) r = await _run_shell(cmd, self.workspace_dir, timeout=timeout, strict=self.generic, secret_env=secret_env) out = f"rc={r['rc']}\n{r['stdout'][:2000]}" if r.get('stderr'): out += f"\nSTDERR: {r['stderr'][:500]}" if self.generic and not r.get('sandbox'): out += "\n（注意：本次未经过沙箱）" # 出站净化（需求2的「执行返回当输入处理」）：工具输出里若回显了 # 已入库凭据明文（报错信息常见），擦洗成占位符再回填模型上下文。 try: from . import secret_vault out, scrubbed = await secret_vault.sanitize_tool_output( sor, out, org_id=self.org_id or "", user_id=self.user_id or "") if scrubbed: out += "\n（输出中出现的敏感信息已替换为占位符：" + ", ".join(scrubbed) + "）" except Exception as e: logger.warning(f"sanitize_tool_output failed: {e}") return out except Exception as e: return f"ERROR: {str(e)[:300]}" async def _t_process(self, sor, p, pid): """后台任务管理入口：先执行，再对输出做出站净化（2026-09-17）。 后台命令可能把凭据回显进 output.log（curl -v、报错信息常见）， 读回模型上下文前擦洗成占位符——与前台 run_command 同一条防线。 """ out = await self._t_process_raw(sor, p, pid) try: from . import secret_vault out, scrubbed = await secret_vault.sanitize_tool_output( sor, out, org_id=self.org_id or "", user_id=self.user_id or "") if scrubbed: out += "\n（输出中出现的敏感信息已替换为占位符：" + ", ".join(scrubbed) + "）" except Exception as e: logger.warning(f"sanitize process output failed: {e}") return out async def _t_process_raw(self, sor, p, pid): """后台任务管理（poll/log/wait/kill，配套 run_command background=true）。 隔离：只解析 self.workspace_dir 下的 .bg/——通用会话 workspace 是 _general/{uid}、项目会话是项目目录，跨用户/跨项目天然不可达； job_id 白名单正则防路径穿越（bg_jobs 内校验）。 """ from . import bg_jobs action = (p.get("action") or "").strip().lower() job_id = (p.get("job_id") or "").strip() if action not in ("poll", "log", "wait", "kill"): return "FAIL: action 须为 poll|log|wait|kill" if not job_id: return "FAIL: 需要 job_id（run_command background=true 的返回值）" try: offset = int(p.get("offset") or 0) except (TypeError, ValueError): offset = 0 try: if action == "poll": r = bg_jobs.poll_job(self.workspace_dir, job_id, offset) m = r["meta"] out = (f"job_id={job_id} status={m['status']} rc={m.get('rc')} " f"sandbox={m.get('sandbox')} log_size={r['log_size']}") if r["new_output"]: out += f"\n--- 新增输出 ---\n{r['new_output'][-4000:]}" return out if action == "log": r = bg_jobs.read_log(self.workspace_dir, job_id, offset) m = r["meta"] more = f"（还有后续，offset={offset + len(r['content'])} 续读）" if r["truncated"] else "" return (f"job_id={job_id} status={m['status']} rc={m.get('rc')}\n" f"--- 输出 ---\n{r['content']}{more}") if action == "wait": r = await bg_jobs.wait_job(self.workspace_dir, job_id) m = r["meta"] out = f"job_id={job_id} status={m['status']} rc={m.get('rc')}" if m["status"] == "running": out += "（等待超时，任务仍在跑，可继续 wait 或 poll）" if r["new_output"]: out += f"\n--- 输出（尾部） ---\n{r['new_output'][-4000:]}" return out if action == "kill": r = bg_jobs.kill_job(self.workspace_dir, job_id) return f"OK: {r['message']}（job_id={job_id}）" except bg_jobs.BgJobError as e: return f"FAIL: {e}" except Exception as e: return f"ERROR: {str(e)[:300]}" return "FAIL: 未知分支" def _resolve_ws_path(self, path: str) -> str: """解析相对路径为工作空间内绝对路径（越界返回 ''）。""" import os ws = self.workspace_dir or WORKSPACE_BASE full = os.path.abspath(os.path.join(ws, path or ".")) # 限制在 workspace 内 if full != ws and not full.startswith(ws.rstrip("/") + "/"): return "" return full async def _t_read_file(self, sor, p, pid): path = p.get("path", "") if not path: return "FAIL: 需要文件路径" full = self._resolve_ws_path(path) if not full: return f"FAIL: 路径越界 {path}" # 统一文件读取（file_read 共享模块，v1/v2 共用）：分页续读 + docx/pdf 解析 + 显式截断告知 from .file_read import read_text_file, DEFAULT_LIMIT try: offset = int(p.get("offset") or 0) except (ValueError, TypeError): offset = 0 r = read_text_file(full, offset=offset, limit=DEFAULT_LIMIT) if r['kind'] == 'error': return f"FAIL: {r['message']} {path}" if r['kind'] == 'binary': return r['message'] return r['content'] async def _t_load_skill(self, sor, p, pid): name = (p.get("name") or "").strip() if not name: return "FAIL: 需要技能名称" if not self._skill_loader: return "FAIL: 技能系统未初始化" merged = self._skill_loader.get_merged( pipeline_id=self.pipeline_id or "", role=self.role or "", project_id=self.project_id or "", org_id=self.org_id or "", user_id=self.user_id or "") from pipeline_core.skill_loader import resolve_skill, format_skill_names skill = resolve_skill(merged, name) if skill is None: names = format_skill_names(merged) return (f"FAIL: 技能 '{name}' 不存在。注意：name 只传技能裸名" f"（如 bid-workflow），不要带 [产线] 前缀或描述文字。可用技能: {names}") file_path = (p.get("file_path") or "").strip() if file_path: body = skill.read_linked_file(file_path) if body.startswith("FAIL") or body.startswith("ERROR"): linked = skill.list_linked_files() return f"{body}\n可用子文件: {', '.join(linked) if linked else '（无）'}" return f"## [{skill.name}] {file_path}\n\n{body}" block = skill.to_prompt_block() linked = skill.list_linked_files() if linked: block += f"\n（可用子文件，用 file_path 参数加载：{', '.join(linked)}）\n" return block async def _t_list_platform_models(self, sor, p, pid): """列出平台可用模型（薄壳委托 platform_model_tools 唯一实现）。""" from .platform_model_tools import tool_list_platform_models return await tool_list_platform_models(p, self.org_id or "0") async def _t_invoke_model(self, sor, p, pid): """调用平台模型完成生成类任务（薄壳委托 platform_model_tools 唯一实现）。""" from .platform_model_tools import tool_invoke_model return await tool_invoke_model( p, self.org_id or "0", user_id=self.user_id or "", project_id=self.project_id or "", session_id=self.session_id or "") async def _t_web_search(self, sor, p, pid): """联网检索（薄壳委托 web_tools 唯一实现，SSRF 防护在其中）。""" from .web_tools import tool_web_search return await tool_web_search(p.get("query", ""), p.get("limit", 8)) async def _t_fetch_url(self, sor, p, pid): """抓取网页正文（超长落盘工作空间 webcache/，read_file 续读）。""" from .web_tools import tool_fetch_url return await tool_fetch_url(p.get("url", ""), workspace_dir=self.workspace_dir or "") async def _t_query_project_data(self, sor, p, pid): """查询项目关联表数据（薄壳委托 db_query 唯一实现，白名单/过滤/审计在其中）。""" if self.generic or not pid: return "FAIL: 当前会话无项目上下文，无法查询项目数据。" from .db_query import tool_query_project_data return await tool_query_project_data( sor, p.get("table", ""), pid, where=p.get("where", ""), order_by=p.get("order_by", ""), limit=p.get("limit", 100), who="agent.main_agent") async def _t_list_packs(self, sor, p, pid): try: from pipeline_core.skill_pack import list_packs packs = list_packs() if not packs: return "（无可安装的技能集）" lines = ["可安装的技能集:"] for pk in packs: lines.append(f" - {pk['name']}: {pk.get('title', '')}（{pk.get('skill_count', 0)} 个技能）") return "\n".join(lines) except Exception as e: return f"ERROR: {str(e)[:300]}" async def _t_install_pack(self, sor, p, pid): pack = (p.get("pack") or "").strip() if not pack: return "FAIL: 需要技能集名（先用 list_packs 查看可用的技能集名）" try: from pipeline_core.skill_pack import install_pack base_dir = self._skill_loader.base_dir if self._skill_loader else "" org_id = self.org_id or "0" r = install_pack(pack, base_dir, org_id) if r.get("success"): if self._skill_loader: self._skill_loader.reload() installed = r.get("installed", []) return f"OK: 已安装技能集 {pack}（{len(installed)} 个技能）" return f"FAIL: {r.get('error', '未知错误')}" except Exception as e: return f"ERROR: {str(e)[:300]}" async def _t_propose_skill(self, sor, p, pid): name = (p.get("name") or "").strip() description = (p.get("description") or "").strip() content = (p.get("content") or "").strip() if not name or not content: return "FAIL: 需要技能名和内容（content 为 SKILL.md 草稿正文）" try: from appPublic.uniqueID import getID # 实时发布（2026-09-08 用户拍板）：写提议者所属机构技能目录 # skills/orgs/{org_id}/{name}/，org scope 同名覆盖 global， # 机构内立即生效、其他机构不受影响；org 0 有用户身份时降级 # users/{user_id}/（只影响本人），org 0 无人值守拒绝（防影响全平台）。 from .skill_live import publish_skill_live ok, msg, info = await publish_skill_live( name, description, content, org_id=self.org_id or "", user_id=self.user_id or "", who=self.user_id or "") await sor.C("skill_proposals", { "id": getID(), "name": name, "description": description, "content": content, "source": "agent", # 实时发布成功=published（审计痕迹）；被拒=保持 pending 人工审核链 "status": "published" if ok else "pending", "feedback": msg if ok else "", "org_id": self.org_id or "", "pipeline_id": self.pipeline_id or "", "created_by": self.user_id or "", }) if ok: return msg # 实时发布被拒（org 0 无人值守等）→ 保持旧行为：提议待审核 return f"OK: 已提交技能提议 '{name}'（{msg}，转人工审核，暂不生效）" except Exception as e: return f"ERROR: {str(e)[:300]}" async def _t_write_file(self, sor, p, pid): path = p.get("path", "") content = p.get("content", "") if not path: return "FAIL: 需要文件路径" full = self._resolve_ws_path(path) if not full: return f"FAIL: 路径越界 {path}" try: os.makedirs(os.path.dirname(full), exist_ok=True) with open(full, "w", encoding="utf-8") as f: f.write(content or "") return f"OK: 已写入 {path} ({len(content)} 字符)" except Exception as e: return f"ERROR: {str(e)[:300]}" async def _t_patch_file(self, sor, p, pid): """定点替换（2026-09-10，对齐 Hermes patch 工具核心语义）。 - old_string 默认须在文件中唯一（0 次/多次都拒绝并给可行动提示）， replace_all=true 时替换全部出现； - 路径经 _resolve_ws_path 越界防护（与 read/write_file 同一边界）； - 二进制/不可解码文件拒绝（文本文件专用）； - 原子写（.tmp + os.replace），防并发读到半截。 """ path = (p.get("path") or "").strip() old_string = p.get("old_string") new_string = p.get("new_string") if not path: return "FAIL: 需要文件路径 path" if not isinstance(old_string, str) or old_string == "": return "FAIL: 需要 old_string（要替换的原文片段，删除时 new_string 传空字符串）" if not isinstance(new_string, str): return "FAIL: 需要 new_string（替换后的文本；删除片段传空字符串）" if old_string == new_string: return "FAIL: old_string 与 new_string 相同，无需修改" full = self._resolve_ws_path(path) if not full: return f"FAIL: 路径越界 {path}" if not os.path.isfile(full): return f"FAIL: 文件不存在 {path}（新建文件用 write_file）" replace_all = str(p.get("replace_all", "")).strip().lower() in ("true", "1", "yes") try: with open(full, "r", encoding="utf-8") as f: text = f.read() except UnicodeDecodeError: return f"FAIL: {path} 不是 UTF-8 文本文件，patch_file 只支持文本" except Exception as e: return f"ERROR: 读取失败 {str(e)[:200]}" n = text.count(old_string) if n == 0: return ("FAIL: old_string 在文件中未找到。先用 read_file 核对原文" "（空白/缩进/换行必须完全一致）") if n > 1 and not replace_all: return (f"FAIL: old_string 在文件中出现 {n} 次（须唯一）。" "请带更多上下文使其唯一，或 replace_all=true 替换全部") new_text = text.replace(old_string, new_string) if replace_all \ else text.replace(old_string, new_string, 1) try: tmp = full + ".patch.tmp" with open(tmp, "w", encoding="utf-8") as f: f.write(new_text) os.replace(tmp, full) except Exception as e: return f"ERROR: 写入失败 {str(e)[:200]}" return (f"OK: {path} 已替换 {n if replace_all else 1} 处" f"（文件 {len(text)}→{len(new_text)} 字符）") async def _t_list_files(self, sor, p, pid): path = p.get("path", "") or "." full = self._resolve_ws_path(path) if not full: return f"FAIL: 路径越界 {path}" try: if not os.path.isdir(full): return f"FAIL: 目录不存在 {path}" items = sorted(os.listdir(full))[:50] lines = [] for name in items: fp = os.path.join(full, name) if name.startswith("."): continue t = "DIR" if os.path.isdir(fp) else "FILE" size = os.path.getsize(fp) if os.path.isfile(fp) else 0 lines.append(f"[{t}] {name} ({size}B)") return "\n".join(lines) if lines else "(空目录)" except Exception as e: return f"ERROR: {str(e)[:300]}" async def _t_search_files(self, sor, p, pid): pattern = p.get("pattern", "") if not pattern: return "FAIL: 需要搜索关键词" path = p.get("path", "") or "." full = self._resolve_ws_path(path) if not full: return f"FAIL: 路径越界 {path}" try: from pipeline_service.agent_loop import _run_shell # grep -rn，排除 .git 和 __pycache__，限制输出 r = await _run_shell( f"grep -rn --include='*.py' --include='*.md' --include='*.json' --include='*.txt' " f"--exclude-dir=.git --exclude-dir=__pycache__ '{pattern}' . 2>/dev/null | head -50", full, timeout=30) out = r.get("stdout", "").strip() return out[:4000] if out else f"未找到匹配 '{pattern}' 的内容" except Exception as e: return f"ERROR: {str(e)[:300]}" async def _t_session_search(self, sor, p, pid): query = p.get("query", "").strip() if not query: return "FAIL: 需要搜索关键词" try: # 产线隔离（2026-09-05 第7层）：通用会话只搜自己的历史 # （pipeline_id=''），与 _load_history 过滤对称；产线会话按 # iteration_id（项目）过滤不变。 if self.generic: sql = ("SELECT role, content, created_at FROM pipeline_conversations " "WHERE created_by=${u}$ AND (pipeline_id='' OR pipeline_id IS NULL) " "AND content LIKE ${q}$ " "ORDER BY created_at DESC LIMIT 10") else: sql = ("SELECT role, content, created_at FROM pipeline_conversations " "WHERE created_by=${u}$ AND iteration_id=${pid}$ AND content LIKE ${q}$ " "ORDER BY created_at DESC LIMIT 10") recs = await sor.sqlExe(sql, {"u": self.user_id, "pid": pid, "q": f"%{query}%"}) if not recs: return f"未找到包含 '{query}' 的会话记录" lines = [] for r in recs: content = (getattr(r, "content", "") or "")[:300] lines.append(f"[{getattr(r, 'role', '?')}] {content}") return "\n".join(lines) except Exception as e: return f"ERROR: {str(e)[:300]}" async def _t_memory(self, sor, p, pid): """持久记忆工具（add/list/remove，2026-09-10 对齐 Hermes memory）。 多租户写入门禁（服务多机构多用户，全部代码层强制，不靠 LLM 自觉）： - org_id/user_id 一律用会话真实身份注入，忽略 LLM 传值（防伪造归属） - scope 白名单：user/project/pipeline——global/org 是平台种子域，禁写 （global 域所有人可见，agent 写入=跨机构泄漏；org 域是平台运营配置） - scope=project 必须有当前项目；scope=pipeline 的 scope_id 强制用 当前产线（防写进他产线域） - remove 只能删「本机构 org_id + 本人或本机构名下」的条目 （store.remove 带 org/user 过滤），种子与他租户条目删不到 - 无 user_id（无人值守场景）拒绝写入——归属不明的记忆一律不收 """ if not self._memory_store: return "FAIL: 记忆系统未初始化" from pipeline_core.memory_store import MemoryStore action = (p.get("action") or "").strip().lower() store = self._memory_store org_id = self.org_id or "" user_id = self.user_id or "" if action == "add": content = (p.get("content") or "").strip() if not content: return "FAIL: 需要 content（记忆内容）" if not user_id: return "FAIL: 无用户身份的会话不允许写记忆（归属不明）" scope = (p.get("scope") or "user").strip().lower() if scope not in MemoryStore.WRITABLE_SCOPES: return ("FAIL: scope 只允许 user/project/pipeline" "（global/org 是平台种子域，agent 禁写）") category = (p.get("category") or "memory").strip().lower() if category not in ("user", "memory"): category = "memory" scope_id = "" if scope == "project": if not self.project_id: return "FAIL: 当前会话没有项目上下文，scope=project 不可用（改用 user 或 pipeline）" scope_id = self.project_id elif scope == "pipeline": if self.generic or not self.pipeline_id: return "FAIL: 通用会话没有产线上下文，scope=pipeline 不可用（改用 user）" scope_id = self.pipeline_id elif scope == "user": scope_id = user_id priority = MemoryStore.PRIORITY_HIGH if category == "user" else MemoryStore.PRIORITY_MEDIUM # user_id 一律记创建者（所有 scope）：user 域据此仅本人可见； # project/pipeline 域可见性仍由 org_id 控制（机构内共享）， # 但删除时按 (org_id+user_id) 过滤=只能删本人写入的， # 防同机构他人误删共享条目 await store.add(content=content, category=category, priority=priority, scope=scope, scope_id=scope_id, org_id=org_id, user_id=user_id) scope_cn = {"user": "个人偏好", "project": "当前项目", "pipeline": "当前产线"}[scope] return f"OK: 已记住（{scope_cn}域，跨会话生效）" if action == "list": entries = await store.get_visible(org_id, user_id, pipeline_id="" if self.generic else self.pipeline_id, project_id=self.project_id) if not entries: return "(暂无可见记忆)" lines = [] for e in entries[:30]: lines.append(f"- [{e.scope}] key={e.key}: {e.content[:80]}") if len(entries) > 30: lines.append(f"（共 {len(entries)} 条，仅显示前 30）") return "\n".join(lines) if action == "remove": key = (p.get("key") or "").strip() if not key: return "FAIL: remove 需要 key（先用 action=list 查）" # 过滤 (org_id + 本人 user_id)：只能删本人写入的条目。 # 种子条目 org_id=''、他机构条目 org_id≠本机构、同机构他人条目 # user_id≠本人，全部够不着；key 掺了租户维度防跨租户误删。 removed = 0 for cat in ("memory", "user"): removed += await store.remove(key, cat, org_id=org_id, user_id=user_id) if removed: return f"OK: 已删除记忆 {key}" return f"FAIL: 未找到可删除的记忆 {key}（只能删本机构/本人的条目）" return "FAIL: action 须为 add|list|remove" async def _t_manage_skill(self, sor, p, pid): """技能管理（create/patch/write_file/remove_file/delete，2026-09-10）。 隔离全部由 skill_live.manage_skill_live 把关：只能落本机构 orgs/{org}/ 或 users/{uid}/ 目录；global 原版走 fork-on-write 继承副本；产线/角色/ 项目层技能拒绝修改；delete 只删本租户副本。org_id/user_id 用会话真实 身份注入，忽略 LLM 传值。 """ from .skill_live import manage_skill_live action = (p.get("action") or "").strip().lower() name = (p.get("name") or "").strip() if not action: return "FAIL: 需要 action（create/patch/write_file/remove_file/delete）" ok, msg, info = await manage_skill_live( action=action, name=name, org_id=self.org_id or "", user_id=self.user_id or "", who=self.user_id or "", description=(p.get("description") or "").strip(), content=p.get("content") or "", old_string=p.get("old_string") or "", new_string=p.get("new_string") or "", file_path=(p.get("file_path") or "").strip(), file_content=p.get("file_content") or "", pipeline_id="" if self.generic else (self.pipeline_id or ""), role=self.role or "", project_id=self.project_id or "", ) if ok: # 审计：技能增改删留痕（append-only） try: from .audit import record_audit await record_audit( tenant_id=self.org_id or "0", entity="skills", entity_id=name, action=f"skill_{action}", who=self.user_id or "agent", detail=str(info)[:500], sor=sor) except Exception as e: logger.warning(f"manage_skill audit failed: {e}") return msg if ok else f"FAIL: {msg}" async def _t_todo(self, sor, p, pid): action = p.get("action", "list") content = (p.get("content", "") or "").strip() if action == "add": if not content: return "FAIL: add 需要任务内容" self._todos.append({"done": False, "content": content}) return f"OK: 已添加任务（共 {len(self._todos)} 项）" if action == "done": if not content: return "FAIL: done 需要任务序号或内容" for t in self._todos: if content in t["content"] or content == str(self._todos.index(t) + 1): t["done"] = True return f"OK: 已完成任务 '{t['content']}'" return f"未找到任务 '{content}'" # list（默认） if not self._todos: return "任务清单为空" lines = [] for i, t in enumerate(self._todos): mark = "✅" if t["done"] else "⏳" lines.append(f"{i + 1}. {mark} {t['content']}") return "\n".join(lines) async def _t_delegate_subtask(self, sor, p, pid): """委派子 agent（2026-09-10 升级，对齐 Hermes delegate_task）。 background=false（默认）：同步等待，行为同旧版（兼容存量用法）。 background=true：后台并行（同 workspace 上限 3 个），立即返回 subagent_id，用 subagent 工具 list/steer/stop/result 跟进。 隔离（两种模式一致）： - 子 executor session_isolation='none'：不读父历史、不写 pipeline_conversations（子任务对话不污染父会话回放） - 子 agent 不继承 session_id（多 tab 项目上下文不串） - 深度限制：子 agent（_delegate_depth≥1）不能再委派，防递归爆炸 """ goal = (p.get("goal", "") or "").strip() context = (p.get("context", "") or "").strip() if not goal: return "FAIL: 需要子任务目标" bg = str(p.get("background", "")).strip().lower() in ("true", "1", "yes") if bg: from . import subagents try: sid = await subagents.spawn_subagent(self, goal, context) except subagents.SubagentError as e: return f"FAIL: {e}" return (f"OK: 子agent已后台启动 subagent_id={sid}。" f"用 subagent(action=list) 查状态、steer 追加指示、" f"stop 终止、result 取结果（完成前 result 是中间输出）。") # 同步模式（深度限制同样生效） from . import subagents if self._delegate_depth >= subagents.MAX_DEPTH: return "FAIL: 子agent不能再委派子任务（深度限制，防递归爆炸）" try: from dataclasses import replace as _dc_replace child_config = _dc_replace(self.config, session_isolation="none") sub = AgentExecutor( config=child_config, project_id=self.project_id, user_id=self.user_id, workspace_dir=self.workspace_dir, model_name=self.model_name, session_id="", generic=self.generic, default_pipeline_id=self.pipeline_id, delegate_depth=self._delegate_depth + 1, ) prompt = goal if not context else f"{goal}\n\n背景：{context}" result_parts = [] async for chunk in sub.run(prompt): try: data = json.loads(chunk) except Exception: continue t = data.get("type", "") if t == "reply": result_parts.append(data.get("message", "")) elif t == "tool_result": result_parts.append(data.get("result", "")) out = "\n".join(x for x in result_parts if x).strip() return out[:3000] or "(子任务无输出)" except Exception as e: return f"ERROR: {str(e)[:300]}" async def _t_subagent(self, sor, p, pid): """后台子agent管理（list/steer/stop/result）。隔离靠 workspace 目录。""" from . import subagents action = (p.get("action") or "").strip().lower() sid = (p.get("subagent_id") or "").strip() try: if action == "list": items = subagents.list_subagents(self.workspace_dir) if not items: return "(本会话还没有派生过后台子agent)" lines = [] for it in items[:10]: lines.append( f"- {it['subagent_id']} [{it['status']}] {it['goal']}" + (f"（错误: {it['error']}）" if it.get("error") else "")) return "\n".join(lines) if not sid: return "FAIL: 需要 subagent_id（先用 action=list 查看）" if action == "steer": return subagents.steer_subagent( self.workspace_dir, sid, (p.get("message") or "").strip()) if action == "stop": return subagents.stop_subagent(self.workspace_dir, sid) if action == "result": r = subagents.get_result(self.workspace_dir, sid) m = r["meta"] return (f"subagent_id={sid} status={m['status']}\n" f"--- 输出 ---\n{r['result']}") except subagents.SubagentError as e: return f"FAIL: {e}" except Exception as e: return f"ERROR: {str(e)[:300]}" return "FAIL: action 须为 list|steer|stop|result" # ═══════════════════════════════════════════════════════ # 上下文压缩 # ═══════════════════════════════════════════════════════ async def _maybe_compress(self): """检查是否需要压缩上下文""" est_tokens = self._estimate_tokens() limit = self.config.context_limit threshold = int(limit * self.config.compression.threshold) if est_tokens < threshold: return logger.info(f"Compressing: {est_tokens}/{limit} tokens (threshold={threshold})") # 保留 system + 最近 N 轮 keep = self.config.compression.keep_recent * 2 # user + assistant 各一条 system_msg = self._msgs[0] if self._msgs else None recent = self._msgs[-keep:] if keep < len(self._msgs) else self._msgs[1:] # 压缩中间消息 middle = self._msgs[1:-keep] if keep < len(self._msgs) else [] if middle: summary = await self._summarize(middle) compressed = [ system_msg, {"role": "user", "content": f"[对话摘要]\n{summary}"}, ] if system_msg else [] compressed.extend(recent) self._msgs = compressed def _estimate_tokens(self) -> int: """简单 token 估算（中文 ~1.5 字符/token，英文 ~4 字符/token）""" total = 0 for m in self._msgs: content = m.get("content") or "" if isinstance(content, list): # 多模态消息（2026-09-10）：文本部分按字符算，图片按固定 1000 token # 估（base64 不进文本计数——图片实际 token 由视觉模型分块决定， # 此处只需量级正确触发压缩阈值） for pt in content: if pt.get("type") == "text": total += len(pt.get("text") or "") // 2 + 1 elif pt.get("type") == "image_url": total += 1000 continue # 简单估算：平均每 2 字符 1 token total += len(content) // 2 + 1 return total @staticmethod def _content_as_text(content) -> str: """消息 content 归一为纯文本（多模态 list 取文本部分 + 图片占位）。 压缩/估算/日志统一走这里——list content 直接 len()/join 会 TypeError （同 native FC content=None 的教训：所有碰 _msgs content 的地方都要防御）。 """ if isinstance(content, list): parts = [] for pt in content: if pt.get("type") == "text": parts.append(pt.get("text") or "") elif pt.get("type") == "image_url": parts.append("[图片]") return "\n".join(parts) return content or "" async def _summarize(self, messages: list) -> str: """压缩消息为摘要""" if len(messages) <= 2: return "\n".join(self._content_as_text(m.get("content"))[:200] for m in messages) text = "\n".join( f"[{m['role']}]: {self._content_as_text(m.get('content'))[:500]}" for m in messages ) try: from pipeline_service.llm_bridge import llm_call summary = await llm_call( f"请用3-5句话总结以下对话的关键信息：\n\n{text[:4000]}", temperature=0.1, model=self.model_name, org_id=self.org_id, purpose='utility', session_id=self.session_id, timeout=self._UTILITY_TIMEOUT, ) return summary[:500] except Exception: return text[:500] # ═══════════════════════════════════════════════════════ # 会话管理 # ═══════════════════════════════════════════════════════ async def _load_history(self, external_history: List[dict] = None) -> List[dict]: """加载历史消息。隔离策略：session_id > project 级别。 多 tab 独立会话：session_id 非空时按 session_id 隔离（每个 tab 独立历史）； 否则退回 config.session_isolation（project/user/none）。 """ if external_history: return external_history if self.config.session_isolation == "none": return [] try: from sqlor.dbpools import DBPools db = DBPools() async with db.sqlorContext("pipeline") as sor: # 会话级隔离（web 多 tab 独立会话） # 注意（2026-09-01 修复）：必须 ORDER BY created_at DESC LIMIT N 取「最新」N 条 # 再倒序回时间正序。历史用 ASC LIMIT 会取「最旧」N 条——会话消息超过 # history_limit 后，最近的提问（如 A/B/C 选项）被截掉，LLM 看不到刚问的 # 问题，按更旧的上下文答非所问（用户答 C，agent 复读 4 小时前的旧回复）。 if self.session_id: # 产线隔离（2026-08-31 修复跨产线串扰）：各产线页面默认 tab 共用 # session_id='default'，历史若不按产线过滤，投标页会加载开发页的 # 对话（如"元景项目/新建demo项目"），LLM 跟着最后上下文答错产线。 # 2026-09-05 第7层泄漏修复：通用会话（存库时 pipeline_id=''）只回放 # 自己的历史，否则会加载该用户各产线的最近对话——项目名称/ID # 等产线信息随历史泄漏进通用助手。 sql = ("SELECT role, content FROM pipeline_conversations " "WHERE session_id=${sid}$ AND created_by=${uid}$ ") params = {"sid": self.session_id, "uid": self.user_id or "", "lim": self.config.history_limit} if self.generic: sql += "AND (pipeline_id='' OR pipeline_id IS NULL) " elif self.pipeline_id: sql += "AND pipeline_id=${pl}$ " params["pl"] = self.pipeline_id sql += "ORDER BY created_at DESC LIMIT ${lim}$" recs = await sor.sqlExe(sql, params) else: # 项目隔离 pid = self.project_id if self.config.session_isolation == "project" and pid: recs = await sor.sqlExe( "SELECT role, content FROM pipeline_conversations " "WHERE iteration_id=${pid}$ AND created_by=${uid}$ " "ORDER BY created_at DESC LIMIT ${lim}$", {"pid": pid, "uid": self.user_id or "", "lim": self.config.history_limit}, ) elif self.generic: # 2026-09-05 第7层泄漏修复：通用会话只回放自己的历史 # （存库时 generic 会话 pipeline_id=''），禁止按 user_id 全量 # 回放——否则用户各产线最近对话（含项目名称/ID）全部泄漏。 recs = await sor.sqlExe( "SELECT role, content FROM pipeline_conversations " "WHERE created_by=${uid}$ AND (pipeline_id='' OR pipeline_id IS NULL) " "ORDER BY created_at DESC LIMIT ${lim}$", {"uid": self.user_id or "", "lim": self.config.history_limit}, ) else: recs = await sor.sqlExe( "SELECT role, content FROM pipeline_conversations " "WHERE created_by=${uid}$ " "ORDER BY created_at DESC LIMIT ${lim}$", {"uid": self.user_id or "", "lim": self.config.history_limit}, ) if not recs: return [] recs = list(reversed(recs)) # DESC 取出 → 倒回时间正序 # 过滤 tool_call JSON 污染（关键！） msgs = [] for r in recs: role = getattr(r, "role", "") content = getattr(r, "content", "") or "" # 跳过 tool_call JSON if content.startswith('{"action":"tool_call"'): continue if content.startswith("已调用 "): continue if role in ("user", "assistant"): msgs.append({"role": role, "content": content}) return msgs except Exception as e: logger.warning(f"History load failed: {e}") return [] async def _save_turn(self, user_input: str, reply: str): """持久化本轮对话""" if self.config.session_isolation == "none": return # PII 落库掩码（2026-09-17）：会话表是长期留存+session_search 可捞，占位符 # 虽非明文但缓存 24h 内可还原 → 落库一律换掩码形态（138****5678）， # 缓存过期后掩码仍是唯一可见形态（合规：留存数据不含可还原 PII）。 if self.pii_session_key and "@@pii:" in (user_input + reply): try: from . import pii_guard if "@@pii:" in user_input: user_input = await pii_guard.restore_outbound( user_input, self.pii_session_key, masked=True) if "@@pii:" in reply: reply = await pii_guard.restore_outbound( reply, self.pii_session_key, masked=True) except Exception as e: logger.warning(f"pii mask on save failed: {e}") try: from sqlor.dbpools import DBPools from appPublic.uniqueID import getID db = DBPools() async with db.sqlorContext("pipeline") as sor: # 产线隔离（2026-08-31）：generic 会话不挂产线标签——generic 的 # pipeline_id 会回退 DEFAULT_ABILITY_ID(sdlc_general)，若落库会污染 # 开发产线历史。存空串 = 通用对话，各产线读取时按产线过滤自然排除。 store_pl = "" if self.generic else (self.pipeline_id or "") # 用户消息 await sor.C("pipeline_conversations", { "id": getID(), "role": "user", "content": user_input, "created_by": self.user_id or "", "iteration_id": self.project_id or "", "session_id": self.session_id or "", "pipeline_id": store_pl, }) # 助手回复（不含 tool_call JSON！） reply_clean = reply if reply.startswith('{"action":"tool_call"'): reply_clean = "[工具调用]" await sor.C("pipeline_conversations", { "id": getID(), "role": "assistant", "content": reply_clean[:4000], "created_by": self.user_id or "", "iteration_id": self.project_id or "", "session_id": self.session_id or "", "pipeline_id": store_pl, }) except Exception as e: logger.warning(f"Session save failed: {e}") async def _load_project_context(self) -> str: """加载项目上下文（项目名、任务数等）""" if not self.project_id: return "" try: from sqlor.dbpools import DBPools db = DBPools() async with db.sqlorContext("pipeline") as sor: recs = await sor.sqlExe( "SELECT name, description FROM sd_projects WHERE id=${pid}$", {"pid": self.project_id}) if recs: name = getattr(recs[0], "name", "") return f"项目: {name}" except Exception: pass return "" def _is_dangerous_command(self, cmd: str) -> bool: """判断 shell 命令是否危险（run_command 只对危险命令要求确认）。""" c = (cmd or "").strip().lower() dangerous = ( "rm -rf", "rm -r", "sudo", "su ", "drop table", "drop database", "truncate", "delete from", "mkfs", "dd if", ":(){", "shutdown", "reboot", "kill -9", "chmod 777", "chown", "> /dev/", ) return any(d in c for d in dangerous) def _detect_confirm_decision(self, user_input: str) -> str: """检测用户对危险命令确认的回复。返回 approve / approve_all / cancel / ''。""" t = (user_input or "").strip().lower() if not t: return "" for kw in ("全部确认", "确认全部", "全部同意", "都确认", "都同意", "一律确认", "approve all", "yes to all", "always"): if kw in t: return "approve_all" for kw in ("取消", "放弃", "不执行", "别执行", "cancel", "abort", "不要"): if kw in t: return "cancel" for kw in ("确认", "同意", "执行", "可以", "approve", "confirm", "yes", "ok", "继续"): if kw in t: return "approve" return "" def _needs_confirmation(self, tool_name: str, params: dict = None) -> bool: """检查工具是否需要用户确认。 会话级「全部确认」（approve_all）时直接放行；否则 run_command 只对 危险命令确认，普通命令自动执行，其他 requires_confirmation 工具一律确认。 别名先归一再查注册表（2026-08-31 修复）：否则 LLM 用别名（如 stop_project） 调需确认工具时，注册表查不到 → 误判无需确认 → 绕过确认门直接执行。 """ tool_name = _PROJECT_TOOL_ALIASES.get(tool_name, tool_name) if self._session and getattr(self._session, "approve_all", False): return False if self._tool_registry: tool = self._tool_registry.get(tool_name) if tool and tool.requires_confirmation: if tool_name == "run_command": cmd = (params or {}).get("command", "") return self._is_dangerous_command(cmd) return True return False # ═══════════════════════════════════════════════════════════ # 便捷函数 # ═══════════════════════════════════════════════════════════ async def run_agent( user_input: str, project_id: str = "", user_id: str = "", config=None, model_name: str = None, workspace_dir: str = "", ) -> AsyncGenerator[str, None]: """便捷入口：运行 agent 并流式输出。 用法： async for chunk in run_agent("创建一个新项目"): print(chunk) """ if config is None: try: from pipeline_core.agent_config import load_agent_config, SDLC_DEFAULT_CONFIG config = await load_agent_config(project_id=project_id) except ImportError: config = SDLC_DEFAULT_CONFIG executor = AgentExecutor( config=config, project_id=project_id, user_id=user_id, workspace_dir=workspace_dir, model_name=model_name, ) async for chunk in executor.run(user_input): yield chunk