From 1c04efe1be495b257cce1bf591d3aa5e4d2e822f Mon Sep 17 00:00:00 2001 From: ymq Date: Thu, 10 Sep 2026 17:09:45 +0800 Subject: [PATCH] =?UTF-8?q?feat(agent):=20=E4=BC=9A=E8=AF=9Dagent=E5=9B=9B?= =?UTF-8?q?=E9=A1=B9Hermes=E8=83=BD=E5=8A=9B=E5=AF=B9=E9=BD=90(=E8=AE=B0?= =?UTF-8?q?=E5=BF=86=E5=86=99=E5=85=A5/=E6=8A=80=E8=83=BD=E7=AE=A1?= =?UTF-8?q?=E7=90=86/=E5=90=8E=E5=8F=B0=E8=BF=9B=E7=A8=8B/=E5=B9=B6?= =?UTF-8?q?=E8=A1=8C=E5=A7=94=E6=B4=BE)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. memory 工具(add/list/remove): 多租户写入门禁——org_id/user_id强制注入、 scope白名单user/project/pipeline(global/org种子域禁写)、无身份拒写、 remove仅本人条目; 记忆注入改可见性过滤版(修跨机构泄漏) 2. manage_skill(create/patch/write_file/remove_file/delete): skill_live扩展, 只落本租户orgs/users目录, global原版fork-on-write(校验通过才fork,失败零残留), org+user双副本同步改, 产线层拒改, delete只删本租户副本+审计留痕 3. run_command background=true + process工具(poll/log/wait/kill): bg_jobs.py 状态文件化(workspace/.bg/,跨worker可见), 沙箱档位与前台一致(generic强制 strict+无bwrap拒绝), 超时SIGTERM进程组+stale心跳判活 4. delegate_subtask background + subagent工具(list/steer/stop/result): subagents.py 并行上限3/workspace, 深度限制1(子agent禁再委派), 子会话session_isolation=none(不读父历史不写会话表), steer/stop文件传递 每轮tool-loop边界消费, 结果流式落盘(stop即部分结果) --- pipeline_service/agent_loop_v2.py | 321 ++++++++++++++++++++++++++- pipeline_service/bg_jobs.py | 273 +++++++++++++++++++++++ pipeline_service/skill_live.py | 294 +++++++++++++++++++++++++ pipeline_service/subagents.py | 350 ++++++++++++++++++++++++++++++ 4 files changed, 1228 insertions(+), 10 deletions(-) create mode 100644 pipeline_service/bg_jobs.py create mode 100644 pipeline_service/subagents.py diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index d98121e..28455a9 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/pipeline_service/agent_loop_v2.py @@ -100,6 +100,7 @@ class AgentExecutor: 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，禁止再委派） ): self.config = config self.project_id = project_id @@ -123,6 +124,9 @@ class AgentExecutor: 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 钩子用） # 懒加载 self._tool_registry = None @@ -226,6 +230,30 @@ class AgentExecutor: for turn in range(max_turns): self._turn_count = turn + 1 + # 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() @@ -526,6 +554,7 @@ class AgentExecutor: temperature=0, org_id=self.org_id, purpose='utility', + project_id=self.project_id or '', ) m = _re.search(r"\[[^\]]*\]", content or "") if m: @@ -554,17 +583,18 @@ class AgentExecutor: # 产线隔离（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, scope="global") + 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=15, scope="pipeline", scope_id=self.pipeline_id) - if self.project_id: - proj_block = await self._memory_store.build_prompt_block( - max_entries=5, scope="project", scope_id=self.project_id) - if proj_block: - mem_block = (mem_block + "\n" + proj_block) if mem_block else proj_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}" @@ -639,6 +669,7 @@ class AgentExecutor: model=self.model_name, temperature=self.config.temperature, org_id=self.org_id, + project_id=self.project_id or '', ) except Exception as e: logger.error(f"native function calling failed, fallback to text: {e}") @@ -650,6 +681,7 @@ class AgentExecutor: model=self.model_name, temperature=self.config.temperature, org_id=self.org_id, + project_id=self.project_id or '', ) return {"content": content or "", "tool_calls": []} @@ -871,6 +903,7 @@ class AgentExecutor: "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, "load_skill": self._t_load_skill, @@ -881,8 +914,11 @@ class AgentExecutor: "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, @@ -1213,7 +1249,27 @@ class AgentExecutor: if self.generic and not _find_bwrap(): return ("FAIL: 通用会话的命令执行需要 bwrap 沙箱（当前服务器不可用），已拒绝执行。" "文件类操作请改用 read_file/write_file/list_files/search_files。") - r = await _run_shell(cmd, self.workspace_dir, timeout=60, strict=self.generic) + + # 后台执行（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) + 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) out = f"rc={r['rc']}\n{r['stdout'][:2000]}" if r.get('stderr'): out += f"\nSTDERR: {r['stderr'][:500]}" @@ -1223,6 +1279,57 @@ class AgentExecutor: except Exception as e: return f"ERROR: {str(e)[:300]}" + async def _t_process(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 @@ -1293,7 +1400,8 @@ class AgentExecutor: """调用平台模型完成生成类任务（薄壳委托 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 "") + p, self.org_id or "0", user_id=self.user_id or "", + project_id=self.project_id or "") async def _t_web_search(self, sor, p, pid): """联网检索（薄壳委托 web_tools 唯一实现，SSRF 防护在其中）。""" @@ -1469,6 +1577,131 @@ class AgentExecutor: 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() @@ -1495,17 +1728,51 @@ class AgentExecutor: 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=self.config, + 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 = [] @@ -1524,6 +1791,40 @@ class AgentExecutor: 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" + # ═══════════════════════════════════════════════════════ # 上下文压缩 # ═══════════════════════════════════════════════════════ diff --git a/pipeline_service/bg_jobs.py b/pipeline_service/bg_jobs.py new file mode 100644 index 0000000..a1c8ece --- /dev/null +++ b/pipeline_service/bg_jobs.py @@ -0,0 +1,273 @@ +# -*- coding: utf-8 -*- +"""bg_jobs.py - run_command 后台任务（2026-09-10，对齐 Hermes terminal background + process_manage） + +状态文件化：{workspace}/.bg/{job_id}/ 下 + meta.json {job_id, command, pid, status(running/done/failed/timeout/killed), + started_at, ended_at, rc, sandbox} + output.log stdout+stderr 合并追加 + +为什么文件化而不是内存注册表：web/worker 多进程（未来多主机 NFS 共享 workspace）， +poll/log 请求可能落到没启动该任务的进程上——文件是唯一跨进程事实源。 +进程句柄注册表（_PROCS）只用于本进程 kill 加速，缺失时回退 meta.json 的 pid。 + +隔离：任务目录在发起者的 workspace 内（项目工作空间 / _general/{uid}）， +process 工具只解析 self.workspace_dir 下的 .bg——跨用户/跨项目天然不可达； +job_id 白名单正则防路径穿越。 + +沙箱：与前台 run_command 完全同一条路径（bwrap 参数同款），generic 会话 +strict 档照旧；无 bwrap 时 generic 拒绝（不降级裸 shell）。 + +兜底：后台任务硬上限 MAX_BG_SECONDS（默认1小时），watcher 超时 SIGTERM 进程组； +worker 崩溃后 meta 残留 running——poll 时用 pid 存活检测标记 stale。 +""" +import asyncio +import json +import logging +import os +import re +import signal +import time + +logger = logging.getLogger("pipeline.bg_jobs") + +BG_DIR = ".bg" +MAX_BG_SECONDS = 3600 +_JOB_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,32}$") + +# 本进程启动的任务句柄（跨进程 kill 回退 pid） +_PROCS = {} +# watcher 协程引用（asyncio.create_task 不保引用会被 GC 中途回收） +_WATCHERS = set() + + +class BgJobError(Exception): + pass + + +def _bg_root(workspace_dir: str) -> str: + return os.path.join(workspace_dir, BG_DIR) + + +def _validate_job_id(job_id: str) -> str: + jid = (job_id or "").strip() + if not _JOB_ID_RE.match(jid): + raise BgJobError(f"非法 job_id: {job_id!r}") + return jid + + +def job_dir(workspace_dir: str, job_id: str) -> str: + jid = _validate_job_id(job_id) + return os.path.join(_bg_root(workspace_dir), jid) + + +def _write_meta(jdir: str, meta: dict): + tmp = os.path.join(jdir, "meta.json.tmp") + with open(tmp, "w", encoding="utf-8") as f: + json.dump(meta, f, ensure_ascii=False) + os.replace(tmp, os.path.join(jdir, "meta.json")) + + +def _read_meta(jdir: str) -> dict: + path = os.path.join(jdir, "meta.json") + if not os.path.exists(path): + raise BgJobError("任务不存在（meta.json 缺失）") + with open(path, "r", encoding="utf-8") as f: + return json.load(f) + + +def _pid_alive(pid: int) -> bool: + if not pid: + return False + try: + os.kill(pid, 0) + return True + except OSError: + return False + + +async def start_bg_job(command: str, workspace_dir: str, strict: bool = False, + timeout: int = MAX_BG_SECONDS) -> str: + """启动后台命令，返回 job_id。沙箱/目录安全校验与前台 _run_shell 完全一致。""" + from .agent_loop import (_find_bwrap, _build_agent_bwrap_cmd, + _sandbox_writable_root, _is_safe_workdir_async) + + command = (command or "").strip() + if not command: + raise BgJobError("命令为空") + cwd = os.path.abspath(workspace_dir) + if not await _is_safe_workdir_async(cwd): + raise BgJobError(f"安全限制：目录 {cwd} 不在允许范围") + if not os.path.isdir(cwd): + raise BgJobError(f"目录不存在: {cwd}") + timeout = max(10, min(int(timeout or MAX_BG_SECONDS), MAX_BG_SECONDS)) + + bwrap = _find_bwrap() + if strict and not bwrap: + raise BgJobError("通用会话的后台命令需要 bwrap 沙箱（当前服务器不可用），已拒绝执行") + + from appPublic.uniqueID import getID + job_id = getID()[:12] + jdir = job_dir(workspace_dir, job_id) + os.makedirs(jdir, exist_ok=True) + log_path = os.path.join(jdir, "output.log") + + if bwrap: + writable_root = cwd if strict else await _sandbox_writable_root(cwd) + cmd = _build_agent_bwrap_cmd(bwrap, cwd, writable_root, command, + include_platform_ro=not strict) + else: + cmd = ["/bin/bash", "-c", command] + + log_f = open(log_path, "ab") + try: + proc = await asyncio.create_subprocess_exec( + *cmd, stdout=log_f, stderr=asyncio.subprocess.STDOUT, + cwd=None if bwrap else cwd, + start_new_session=True, # 独立进程组，kill 时整组终止 + ) + finally: + log_f.close() + + meta = { + "job_id": job_id, "command": command[:2000], "pid": proc.pid, + "status": "running", "started_at": time.time(), "ended_at": None, + "rc": None, "sandbox": bool(bwrap), "timeout": timeout, + "strict": bool(strict), + } + _write_meta(jdir, meta) + _PROCS[job_id] = proc + + w = asyncio.create_task(_watch(job_id, jdir, proc, timeout)) + _WATCHERS.add(w) + w.add_done_callback(_WATCHERS.discard) + return job_id + + +async def _watch(job_id: str, jdir: str, proc: asyncio.subprocess.Process, + timeout: int): + """等待进程结束写 rc；超时 SIGTERM 进程组。worker 崩溃则本协程消失， + meta 残留 running，由 poll 的 pid 存活检测兜底标记。""" + status, rc = "done", None + try: + rc = await asyncio.wait_for(proc.wait(), timeout=timeout) + if rc != 0: + status = "failed" + except asyncio.TimeoutError: + status = "timeout" + _terminate(proc) + try: + await asyncio.wait_for(proc.wait(), timeout=10) + except Exception: + pass + except asyncio.CancelledError: + status = "killed" + raise + except Exception as e: + logger.error(f"bg watch {job_id} error: {e}") + status = "failed" + finally: + try: + meta = _read_meta(jdir) + if meta.get("status") == "running": # kill_job 可能已改写 + meta.update({"status": status, "rc": rc, "ended_at": time.time()}) + _write_meta(jdir, meta) + except Exception: + pass + _PROCS.pop(job_id, None) + + +def _terminate(proc): + try: + os.killpg(os.getpgid(proc.pid), signal.SIGTERM) + except Exception: + try: + proc.terminate() + except Exception: + pass + + +def poll_job(workspace_dir: str, job_id: str, offset: int = 0) -> dict: + """查状态 + 新增输出（offset 之后的部分）。跨进程安全（纯文件读）。""" + jdir = job_dir(workspace_dir, job_id) + meta = _read_meta(jdir) + # stale 检测：meta 说 running 但 pid 已不在（启动它的 worker 崩了） + if meta.get("status") == "running" and not _pid_alive(meta.get("pid") or 0) \ + and job_id not in _PROCS: + # 宽限 5 秒（进程刚退出、watcher 还没写 meta 的窗口） + if time.time() - (meta.get("started_at") or 0) > 5: + meta.update({"status": "failed", "rc": None, + "ended_at": time.time()}) + try: + _write_meta(jdir, meta) + except Exception: + pass + new_output = "" + log_path = os.path.join(jdir, "output.log") + total = 0 + if os.path.exists(log_path): + total = os.path.getsize(log_path) + if offset < total: + with open(log_path, "r", encoding="utf-8", errors="replace") as f: + f.seek(offset) + new_output = f.read()[-8000:] + return {"meta": meta, "new_output": new_output, "log_size": total} + + +def read_log(workspace_dir: str, job_id: str, offset: int = 0, + limit: int = 20000) -> dict: + jdir = job_dir(workspace_dir, job_id) + meta = _read_meta(jdir) + log_path = os.path.join(jdir, "output.log") + content, total = "", 0 + if os.path.exists(log_path): + total = os.path.getsize(log_path) + with open(log_path, "r", encoding="utf-8", errors="replace") as f: + if offset: + f.seek(offset) + content = f.read(limit) + return {"meta": meta, "content": content, "offset": offset, + "log_size": total, "truncated": offset + len(content.encode('utf-8', 'replace')) < total} + + +async def wait_job(workspace_dir: str, job_id: str, max_wait: int = 120) -> dict: + """阻塞等待结束（最多 max_wait 秒），超时返回当前状态+部分输出。""" + jdir = job_dir(workspace_dir, job_id) + deadline = time.time() + min(max_wait, 120) + while time.time() < deadline: + meta = _read_meta(jdir) + if meta.get("status") != "running": + return poll_job(workspace_dir, job_id) + proc = _PROCS.get(job_id) + if proc is None and not _pid_alive(meta.get("pid") or 0): + await asyncio.sleep(0.5) # 让 watcher 落 meta + return poll_job(workspace_dir, job_id) + await asyncio.sleep(1) + return poll_job(workspace_dir, job_id) + + +def kill_job(workspace_dir: str, job_id: str) -> dict: + jdir = job_dir(workspace_dir, job_id) + meta = _read_meta(jdir) + if meta.get("status") != "running": + return {"meta": meta, "message": f"任务已结束（{meta.get('status')}），无需终止"} + proc = _PROCS.get(job_id) + if proc is not None: + _terminate(proc) + else: + # 跨进程：按 meta 的 pid 杀进程组 + pid = meta.get("pid") or 0 + if _pid_alive(pid): + try: + os.killpg(os.getpgid(pid), signal.SIGTERM) + except Exception: + try: + os.kill(pid, signal.SIGTERM) + except Exception as e: + raise BgJobError(f"终止失败: {e}") + else: + meta.update({"status": "failed", "ended_at": time.time()}) + _write_meta(jdir, meta) + return {"meta": meta, "message": "进程已不在（可能随启动进程退出），状态已标记"} + meta.update({"status": "killed", "ended_at": time.time()}) + _write_meta(jdir, meta) + return {"meta": meta, "message": "已发送终止信号"} diff --git a/pipeline_service/skill_live.py b/pipeline_service/skill_live.py index 9ba8d13..79c2109 100644 --- a/pipeline_service/skill_live.py +++ b/pipeline_service/skill_live.py @@ -134,3 +134,297 @@ async def publish_skill_live(name: str, description: str, content: str, '本机构' if scope == 'org' else '仅你本人', '' if reloaded else '（loader 刷新失败，下轮会话生效）')) return True, msg, {'scope': scope, 'path': rel} + + +# ═══════════════════════════════════════════════════════════ +# manage_skill_live（2026-09-10，对齐 Hermes skill_manage： +# create/patch/write_file/remove_file/delete） +# +# 隔离铁律（多机构多用户）： +# - 一切写操作只落在本租户目录：机构 orgs/{org}/（org≠0）或 users/{uid}/ +# （org 0 降级 / 个人技能）；global、pipelines（产线/角色/项目层）、 +# 他机构 orgs/ 目录物理不可达（realpath 双检 + 白名单）； +# - patch/write_file/remove_file 的目标若是 global 原版技能 → +# fork-on-write：整目录继承拷贝到本租户目录再改（同名覆盖，只影响本租户）； +# 目标已是本租户既有副本（org 或 user 层）→ 原地改，不迁移不换层； +# 目标是产线/角色/项目层技能 → 拒绝（平台维护，agent 不可改）； +# - delete 只删本租户副本（org + user 两处都清）；global 原版删不到， +# 删掉副本后原版重新对本租户可见； +# - 子文件仅限 references/scripts/templates/assets 四目录（与 +# skill_loader.read_linked_file 白名单一致），路径穿越双检。 +# ═══════════════════════════════════════════════════════════ + +_LINKED_DIRS = ('references', 'scripts', 'templates', 'assets') + + +def _reload_loader(): + try: + from pipeline_core.skill_loader import get_skill_loader + from pipeline_core.skill_pack import get_skills_base + get_skill_loader(get_skills_base()).reload() + return True + except Exception as e: + logger.warning('skill loader reload failed: %s', e) + return False + + +def _check_linked_path(rel_path: str) -> str: + """子文件路径白名单校验（references/scripts/templates/assets），防穿越。""" + rel = (rel_path or '').strip().lstrip('/') + if not rel or '..' in rel.split('/'): + raise ValueError('非法子文件路径: %r' % rel_path) + first = rel.split('/')[0] + if first not in _LINKED_DIRS: + raise ValueError('子文件只允许放在 %s 目录下（收到 %s）' + % ('/'.join(_LINKED_DIRS), rel)) + return rel + + +def _tenant_dirs(org_id: str, user_id: str, name: str, base: str) -> list: + """本租户可写的技能目录候选（org 层在前 = 主目标）。名称非法返回 []。""" + safe_name = _sanitize_id(name) + if not _NAME_RE.match(safe_name): + return [] + dirs = [] + org = _sanitize_id(org_id) + uid = _sanitize_id(user_id) + if org and org != '0': + dirs.append(os.path.join(base, 'orgs', org, safe_name)) + if uid: + dirs.append(os.path.join(base, 'users', uid, safe_name)) + return dirs + + +def _find_existing_skill(name, org_id, user_id, pipeline_id='', role='', project_id=''): + """在租户可见范围内找已存在的技能（返回 Skill 对象或 None）。 + + 用 loader 的 get_merged + resolve_skill：本租户副本（org/user）优先， + 产线/角色/项目层次之，global 原版兜底（供 fork-on-write 继承拷贝）。 + """ + try: + from pipeline_core.skill_loader import get_skill_loader, resolve_skill + from pipeline_core.skill_pack import get_skills_base + loader = get_skill_loader(get_skills_base()) + merged = loader.get_merged(pipeline_id=pipeline_id or '', role=role or '', + project_id=project_id or '', org_id=org_id or '', + user_id=user_id or '') + return resolve_skill(merged, name) + except Exception as e: + logger.warning('find existing skill %s failed: %s', name, e) + return None + + +def _existing_in_tenant(existing, tenant_dirs: list) -> list: + """本租户已存在的技能副本目录列表（loader 生效优先序：user 层在前）。 + + org 层与 user 层可能同时存在同名副本（user 遮蔽 org）——修改必须 + 应用到所有副本，否则改了被遮蔽的那份等于没改。 + """ + dirs = [d for d in tenant_dirs if os.path.isdir(d)] + src_file = getattr(existing, 'path', '') or '' + if src_file: + src = os.path.realpath(os.path.dirname(src_file)) + # loader 实际生效的副本排最前（结果汇报以此为准） + dirs.sort(key=lambda d: 0 if os.path.realpath(d) == src else 1) + return dirs + + +def _assert_inside_tenant(path: str, tenant_dir: str): + """realpath 双检：写路径必须在本租户技能目录内（防 symlink/穿越逃逸）。""" + real = os.path.realpath(path) + root = os.path.realpath(tenant_dir) + if not (real == root or real.startswith(root + os.sep)): + raise ValueError('路径越界（%s 不在租户目录 %s 内），已拒绝' % (path, tenant_dir)) + + +def _atomic_write_text(path: str, text: str): + tmp = path + '.tmp' + with open(tmp, 'w', encoding='utf-8') as f: + f.write(text) + os.replace(tmp, path) + + +async def manage_skill_live(action: str, name: str, org_id: str = '', user_id: str = '', + who: str = '', description: str = '', content: str = '', + old_string: str = '', new_string: str = '', + file_path: str = '', file_content: str = '', + pipeline_id: str = '', role: str = '', project_id: str = ''): + """技能增改删（多租户隔离版）。返回 (ok, message, info)。 + + action: + create 新建/整篇覆盖本租户技能（等价 propose_skill，走 publish_skill_live） + patch 定向替换片段（old_string 须在目标文件唯一；SKILL.md 或子文件） + write_file 写子文件（references/scripts/templates/assets） + remove_file 删子文件 + delete 删除本租户技能副本（global 原版不受影响，删后重新可见） + """ + from pipeline_core.skill_pack import get_skills_base + base = get_skills_base() + action = (action or '').strip().lower() + name = (name or '').strip() + if action not in ('create', 'patch', 'write_file', 'remove_file', 'delete'): + return False, '未知 action: %r（支持 create/patch/write_file/remove_file/delete）' % action, {} + if not name: + return False, '需要技能名 name', {} + + tenant_dirs = _tenant_dirs(org_id, user_id, name, base) + if not tenant_dirs: + # 名称非法，或无 org/user 身份（无人值守 org 0）——与 publish_skill_live 同门禁 + scope, target = resolve_target(org_id, user_id, name) + if not scope: + return False, target, {} + return False, '技能名非法（只允许字母数字._-，且以字母数字开头，≤64字符）', {} + + try: + if action == 'create': + content = (content or '').strip() + if not content: + return False, 'create 需要 content（SKILL.md 正文）', {} + return await publish_skill_live( + name, description, content, org_id=org_id, user_id=user_id, who=who) + + existing = _find_existing_skill(name, org_id, user_id, + pipeline_id, role, project_id) + if existing is None: + return False, ('技能 %r 不存在。patch/write_file/remove_file/delete ' + '只能操作已存在的技能；新建请用 action=create' % name), {} + + if action == 'delete': + import shutil + removed = [] + for d in tenant_dirs: + if os.path.isdir(d): + _assert_inside_tenant(d, os.path.dirname(d)) + shutil.rmtree(d) + removed.append(os.path.relpath(d, base)) + if not removed: + ex_scope = getattr(existing, 'scope', '?') + if ex_scope == 'global': + return False, ('%r 是通用(global)技能，本租户没有副本，平台技能不可删除。' + '如想改掉它的行为，用 action=create 建同名机构副本覆盖' % name), {} + return False, ('%r 位于 %s 层（平台维护），本租户目录下无副本，不可删除' + % (name, ex_scope)), {} + _reload_loader() + return True, ('OK: 已删除本租户技能副本 %s。若存在同名通用(global)技能，' + '它重新对本租户可见' % ', '.join(removed)), \ + {'scope': 'tenant', 'path': removed[0], 'removed': removed} + + # patch / write_file / remove_file：定位操作目录。 + # 本租户可能同时存在 org 层与 user 层副本（user 遮蔽 org）—— + # target_dirs 收集全部需要一致修改的目录，防「改了被遮蔽的那份」。 + # + # fork 延迟到校验通过后（2026-09-10 review 修复）：先 fork 后校验 + # 会在失败路径留下副本残留（如 old_string 未找到仍拷贝了目录）， + # 残留副本遮蔽 global 原版、还会让 delete 误判「有副本可删」。 + # read_dirs = 校验读取目录（副本 or global 原版）； + # ensure_write_dirs() = 校验全过后才 fork 并返回落盘目录。 + target_dirs = _existing_in_tenant(existing, tenant_dirs) + forked = False + fork_note = '' + need_fork = False + if target_dirs: + read_dirs = target_dirs + elif getattr(existing, 'scope', '') == 'global': + need_fork = True + read_dirs = [os.path.realpath(os.path.dirname(getattr(existing, 'path', '')))] + fork_note = ('（原版属 global 层，已继承副本到本租户目录后修改，' + '同名覆盖仅本租户生效、其他机构不受影响）') + else: + return False, ('%r 位于 %s 层（平台/产线维护），agent 不可修改。' + '可修改的范围：本机构(org)/本人(user)副本与通用(global)技能' + '（global 修改走继承副本）' % (name, existing.scope)), {} + + def ensure_write_dirs(): + """校验通过后落盘目录就位：需要 fork 时才拷贝 global 原版。""" + nonlocal target_dirs, forked + if not need_fork: + return target_dirs + import shutil + skill_dir = tenant_dirs[0] + _assert_inside_tenant(skill_dir, os.path.dirname(skill_dir)) + os.makedirs(os.path.dirname(skill_dir), exist_ok=True) + if not os.path.exists(skill_dir): + shutil.copytree(read_dirs[0], skill_dir) + target_dirs = [skill_dir] + forked = True + return target_dirs + + for d in read_dirs + tenant_dirs: + _assert_inside_tenant(d, os.path.dirname(d) if d in tenant_dirs else d) + + if action == 'patch': + rel = _check_linked_path(file_path) if file_path else 'SKILL.md' + if old_string == '': + return False, 'patch 需要 old_string（要替换的原文片段）', {} + # 第一阶段：只读校验（任一读取目录找不到/不唯一即整体拒绝，零落盘） + for d in read_dirs: + fpath = os.path.join(d, rel) + if not os.path.isfile(fpath): + return False, '目标文件不存在: %s（%s，技能内文件用 load_skill 查看）' % (rel, os.path.relpath(d, base)), {} + with open(fpath, 'r', encoding='utf-8') as f: + text = f.read() + n = text.count(old_string) + if n == 0: + return False, 'patch 失败：old_string 在 %s 中未找到（先用 load_skill 核对原文，注意空白/缩进完全一致）' % rel, {} + if n > 1: + return False, 'patch 失败：old_string 在 %s 中出现 %d 次（须唯一，请带更多上下文）' % (rel, n), {} + # 第二阶段：校验全过，落盘目录就位（此时才 fork），同构路径逐个写 + write_dirs = ensure_write_dirs() + texts = [] + for d in write_dirs: + fpath = os.path.join(d, rel) + _assert_inside_tenant(fpath, d) + if not os.path.isfile(fpath): + return False, '目标文件不存在: %s（%s）' % (rel, os.path.relpath(d, base)), {} + with open(fpath, 'r', encoding='utf-8') as f: + text = f.read() + texts.append((fpath, text.replace(old_string, new_string or '', 1))) + for fpath, text in texts: + _atomic_write_text(fpath, text) + _reload_loader() + n_dirs = len(texts) + multi = '（org+user 两层副本已同步修改）' if n_dirs > 1 else '' + return True, 'OK: 已修改 %r 的 %s%s%s，本租户内立即生效' % (name, rel, multi, fork_note), \ + {'scope': 'tenant', 'path': os.path.relpath(texts[0][0], base), + 'paths': [os.path.relpath(t[0], base) for t in texts], 'forked': forked} + + if action == 'write_file': + rel = _check_linked_path(file_path) + if not file_content: + return False, 'write_file 需要 file_content', {} + written = [] + for d in ensure_write_dirs(): + fpath = os.path.join(d, rel) + _assert_inside_tenant(fpath, d) + os.makedirs(os.path.dirname(fpath), exist_ok=True) + _atomic_write_text(fpath, file_content) + written.append(os.path.relpath(fpath, base)) + _reload_loader() + multi = '（org+user 两层副本已同步写入）' if len(written) > 1 else '' + return True, 'OK: 已写入 %r 的子文件 %s%s%s，本租户内立即生效' % (name, rel, multi, fork_note), \ + {'scope': 'tenant', 'path': written[0], 'paths': written, 'forked': forked} + + if action == 'remove_file': + rel = _check_linked_path(file_path) + # 先校验读取目录里至少一处存在该文件（否则不 fork、直接拒绝） + if not any(os.path.isfile(os.path.join(d, rel)) for d in read_dirs): + return False, '子文件不存在: %s' % rel, {} + removed_files = [] + for d in ensure_write_dirs(): + fpath = os.path.join(d, rel) + _assert_inside_tenant(fpath, d) + if os.path.isfile(fpath): + os.remove(fpath) + removed_files.append(os.path.relpath(fpath, base)) + if not removed_files: + return False, '子文件不存在: %s' % rel, {} + _reload_loader() + return True, 'OK: 已删除 %r 的子文件 %s（%d 处副本）' % (name, rel, len(removed_files)), \ + {'scope': 'tenant', 'path': removed_files[0], 'paths': removed_files} + + except ValueError as e: + return False, str(e), {} + except Exception as e: + logger.exception('manage_skill_live %s %s failed', action, name) + return False, 'ERROR: %s' % str(e)[:300], {} + return False, '未实现的动作分支', {} diff --git a/pipeline_service/subagents.py b/pipeline_service/subagents.py new file mode 100644 index 0000000..f197b8d --- /dev/null +++ b/pipeline_service/subagents.py @@ -0,0 +1,350 @@ +# -*- coding: utf-8 -*- +"""subagents.py - 后台并行子 agent 委派（2026-09-10，对齐 Hermes delegate_task） + +对齐能力：spawn（后台并行，同会话上限 MAX_CONCURRENT=3）/ list / steer（中途追加指示） +/ stop（提前终止，返回部分结果）/ result（取最终结果）。 + +状态文件化：{workspace}/.sub/{subagent_id}/ + meta.json {subagent_id, goal, status(running/done/failed/stopped), started_at, + updated_at(心跳), ended_at, depth, parent_session} + result.txt 子 agent 的最终输出（流式追加，stop 时即部分结果） + steer.txt 待消费的追加指示（父 agent 写，子 agent 每轮消费后清空） + stop.flag 存在即要求停止（子 agent 每轮检测） + +文件化的理由与 bg_jobs 相同：web/worker 多进程 + NFS 共享 workspace， +list/steer/stop/result 请求可能落到没跑该子 agent 的进程——文件是唯一跨进程事实源。 +steer/stop 通过文件传递，由运行中的子 agent 在下一轮 tool-loop 边界消费。 + +隔离： +- 目录在发起者 workspace 内（项目工作空间 / _general/{uid}），跨用户/跨项目不可达 +- 并发上限按「同一 workspace」计数，机构 A 的子 agent 不占机构 B 的额度 +- 子 agent 与父会话历史完全隔离：child config.session_isolation='none' + （不读父历史、不写 pipeline_conversations——否则子任务对话污染父会话回放） +- 深度限制：子 agent 不能再委派（MAX_DEPTH=1，防递归爆炸烧钱） +- 心跳 stale 检测：meta.updated_at 超 STALE_SECONDS 仍 running = 启动它的 + 进程已死，标记 failed（不假装还在跑） +""" +import asyncio +import json +import logging +import os +import re +import time +from dataclasses import replace +from typing import Dict, List, Optional + +logger = logging.getLogger("pipeline.subagents") + +SUB_DIR = ".sub" +MAX_CONCURRENT = 3 # 同 workspace 并行上限 +MAX_DEPTH = 1 # 父=0，子=1；子 agent 禁止再委派 +STALE_SECONDS = 300 # 心跳超时判死 +RESULT_LIMIT = 20000 # result.txt 回传上限 +_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,32}$") + +# 本进程正在跑的子 agent：{subagent_id: asyncio.Task} +_TASKS: Dict[str, asyncio.Task] = {} + + +class SubagentError(Exception): + pass + + +def _validate_id(sid: str) -> str: + sid = (sid or "").strip() + if not _ID_RE.match(sid): + raise SubagentError(f"非法 subagent_id: {sid!r}") + return sid + + +def _sub_root(workspace_dir: str) -> str: + return os.path.join(workspace_dir, SUB_DIR) + + +def sub_dir(workspace_dir: str, sid: str) -> str: + return os.path.join(_sub_root(workspace_dir), _validate_id(sid)) + + +def _atomic_write(path: str, text: str): + tmp = path + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: + f.write(text) + os.replace(tmp, path) + + +def _read_meta(sdir: str) -> dict: + path = os.path.join(sdir, "meta.json") + if not os.path.exists(path): + raise SubagentError("子agent不存在（meta.json 缺失）") + with open(path, "r", encoding="utf-8") as f: + return json.load(f) + + +def _write_meta(sdir: str, meta: dict): + meta["updated_at"] = time.time() + _atomic_write(os.path.join(sdir, "meta.json"), + json.dumps(meta, ensure_ascii=False)) + + +def _stale_guard(meta: dict) -> dict: + """running 但心跳超时 → 标记 failed（启动进程已死，不假装还在跑）。""" + if meta.get("status") != "running": + return meta + if time.time() - (meta.get("updated_at") or 0) > STALE_SECONDS: + meta.update({"status": "failed", "ended_at": time.time(), + "error": "心跳超时：启动该子agent的进程可能已退出"}) + return meta + + +# ── 子 agent 运行时钩子（AgentExecutor 每轮调用）── + +def heartbeat(workspace_dir: str, sid: str): + """子 agent 每轮心跳 + 返回待消费指示/停止信号。 + + 返回 (steer_messages: List[str], stop_requested: bool)。 + 文件不存在/解析失败一律当作「无指示、不停止」——钩子绝不能让子 agent 崩。 + """ + steer_msgs: List[str] = [] + stop = False + try: + sdir = sub_dir(workspace_dir, sid) + meta = _read_meta(sdir) + _write_meta(sdir, meta) # 心跳 + steer_path = os.path.join(sdir, "steer.txt") + if os.path.exists(steer_path): + with open(steer_path, "r", encoding="utf-8") as f: + raw = f.read().strip() + if raw: + steer_msgs = [ln for ln in raw.split("\n@@\n") if ln.strip()] + os.remove(steer_path) + if os.path.exists(os.path.join(sdir, "stop.flag")): + stop = True + except Exception as e: + logger.warning(f"subagent heartbeat {sid} failed: {e}") + return steer_msgs, stop + + +def append_result(workspace_dir: str, sid: str, text: str): + """流式追加子 agent 输出（stop 时即为部分结果）。""" + try: + path = os.path.join(sub_dir(workspace_dir, sid), "result.txt") + with open(path, "a", encoding="utf-8") as f: + f.write(text) + except Exception as e: + logger.warning(f"subagent append_result {sid} failed: {e}") + + +def finish(workspace_dir: str, sid: str, status: str, error: str = ""): + try: + sdir = sub_dir(workspace_dir, sid) + meta = _read_meta(sdir) + meta.update({"status": status, "ended_at": time.time()}) + if error: + meta["error"] = error[:500] + _write_meta(sdir, meta) + except Exception as e: + logger.warning(f"subagent finish {sid} failed: {e}") + _TASKS.pop(sid, None) + + +# ── 父 agent 侧操作 ── + +def count_running(workspace_dir: str) -> int: + """本 workspace 下 running 子 agent 数（stale 的不计）。""" + root = _sub_root(workspace_dir) + if not os.path.isdir(root): + return 0 + n = 0 + for name in os.listdir(root): + try: + meta = _stale_guard(_read_meta(os.path.join(root, name))) + if meta.get("status") == "running": + n += 1 + except Exception: + continue + return n + + +async def spawn_subagent(parent_executor, goal: str, context: str = "") -> str: + """后台派生子 agent，返回 subagent_id。 + + parent_executor: 发起委派的 AgentExecutor（继承 user/org/project/pipeline/workspace/config） + 子 agent：独立上下文（不读父历史、不写会话表）、depth+1、禁止再委派。 + """ + goal = (goal or "").strip() + if not goal: + raise SubagentError("需要子任务目标 goal") + + depth = getattr(parent_executor, "_delegate_depth", 0) + if depth >= MAX_DEPTH: + raise SubagentError("子agent不能再委派子任务（深度限制，防递归爆炸）") + + running = count_running(parent_executor.workspace_dir) + if running >= MAX_CONCURRENT: + raise SubagentError(f"并行子agent已达上限 {MAX_CONCURRENT}（当前 {running} 个在跑），" + f"先用 subagent(action=list) 查看、等待或 stop 后再派") + + from appPublic.uniqueID import getID + sid = getID()[:12] + sdir = sub_dir(parent_executor.workspace_dir, sid) + os.makedirs(sdir, exist_ok=True) + _atomic_write(os.path.join(sdir, "result.txt"), "") + _write_meta(sdir, { + "subagent_id": sid, + "goal": goal[:2000], + "context": (context or "")[:2000], + "status": "running", + "depth": depth + 1, + "started_at": time.time(), + "ended_at": None, + "parent_session": getattr(parent_executor, "session_id", "") or "", + "parent_user": getattr(parent_executor, "user_id", "") or "", + }) + + # 子 executor：同 config 但会话隔离关掉（不读父历史、不写 pipeline_conversations） + from .agent_loop_v2 import AgentExecutor + child_config = replace(parent_executor.config, session_isolation="none") + child = AgentExecutor( + config=child_config, + project_id=parent_executor.project_id, + user_id=parent_executor.user_id, + workspace_dir=parent_executor.workspace_dir, + model_name=parent_executor.model_name, + role=getattr(parent_executor, "role", "") or "", + session_id="", # 不继承父 session（历史/项目上下文都不串） + generic=getattr(parent_executor, "generic", False), + default_pipeline_id=getattr(parent_executor, "pipeline_id", "") or "", + delegate_depth=depth + 1, # 子agent禁止再委派 + ) + child._subagent_id = sid # 运行时钩子按此心跳/消费 steer + + prompt = goal if not context else f"{goal}\n\n背景信息：\n{context}" + + async def _runner(): + parts: List[str] = [] + + def _flush(): + # 流式落盘：stop/崩溃时 result.txt 即为部分结果（绝不空手） + _atomic_write(os.path.join(sdir, "result.txt"), + "\n".join(x for x in parts if x)[-RESULT_LIMIT:]) + + try: + async for chunk in child.run(prompt): + try: + data = json.loads(chunk) + except Exception: + continue + t = data.get("type", "") + if t == "reply": + parts.append(data.get("message", "")) + elif t == "tool_result": + # 工具结果摘要（截断），保证中途 stop 也有实质产出可查 + parts.append("[工具 %s] %s" % ( + data.get("tool", "?"), + (data.get("result", "") or "")[:500])) + elif t == "error": + parts.append(f"[错误] {data.get('message', '')}") + elif t == "ask_user": + # 子 agent 无人可问：记录后终止（父 agent 负责澄清） + parts.append(f"[子agent需要澄清，已终止] {data.get('message', '')}") + _flush() + break + _flush() + stopped = os.path.exists(os.path.join(sdir, "stop.flag")) + if stopped: + parts.append("[子任务被父agent提前终止，以上为部分结果]") + _flush() + finish(parent_executor.workspace_dir, sid, + "stopped" if stopped else "done") + except asyncio.CancelledError: + parts.append("[子任务被取消，以上为部分结果]") + try: + _flush() + finish(parent_executor.workspace_dir, sid, "stopped") + finally: + raise + except Exception as e: + logger.exception(f"subagent {sid} crashed") + parts.append(f"\n[异常] {str(e)[:300]}") + try: + _flush() + finally: + finish(parent_executor.workspace_dir, sid, "failed", str(e)) + + _TASKS[sid] = asyncio.create_task(_runner()) + return sid + + +def list_subagents(workspace_dir: str) -> List[dict]: + root = _sub_root(workspace_dir) + if not os.path.isdir(root): + return [] + out = [] + for name in sorted(os.listdir(root)): + try: + sdir = os.path.join(root, name) + meta = _stale_guard(_read_meta(sdir)) + if meta.get("status") != _read_meta(sdir).get("status"): + _write_meta(sdir, meta) # 落盘 stale 判定 + rpath = os.path.join(sdir, "result.txt") + size = os.path.getsize(rpath) if os.path.exists(rpath) else 0 + out.append({ + "subagent_id": meta.get("subagent_id", name), + "goal": (meta.get("goal") or "")[:120], + "status": meta.get("status"), + "started_at": meta.get("started_at"), + "ended_at": meta.get("ended_at"), + "result_bytes": size, + "error": meta.get("error", ""), + }) + except Exception: + continue + out.sort(key=lambda d: d.get("started_at") or 0, reverse=True) + return out + + +def steer_subagent(workspace_dir: str, sid: str, message: str) -> str: + message = (message or "").strip() + if not message: + raise SubagentError("需要追加指示 message") + sdir = sub_dir(workspace_dir, sid) + meta = _stale_guard(_read_meta(sdir)) + if meta.get("status") != "running": + raise SubagentError(f"子agent已{meta.get('status')}，不能追加指示") + path = os.path.join(sdir, "steer.txt") + prev = "" + if os.path.exists(path): + with open(path, "r", encoding="utf-8") as f: + prev = f.read().strip() + merged = f"{prev}\n@@\n{message}" if prev else message + _atomic_write(path, merged) + return "OK: 指示已排队，子agent下一轮消费" + + +def stop_subagent(workspace_dir: str, sid: str) -> str: + sdir = sub_dir(workspace_dir, sid) + meta = _stale_guard(_read_meta(sdir)) + if meta.get("status") != "running": + return f"子agent已{meta.get('status')}，无需停止" + _atomic_write(os.path.join(sdir, "stop.flag"), str(time.time())) + task = _TASKS.get(sid) + if task is not None and not task.done(): + # 本进程：给 runner 一个取消机会（stop.flag 已由 run 循环优雅处理， + # 这里只在 LLM 长调用卡住时兜底） + async def _soft_cancel(): + await asyncio.sleep(90) + if not task.done(): + task.cancel() + asyncio.create_task(_soft_cancel()) + return "OK: 已发送停止信号（子agent在当前轮结束后停止，已有部分结果保留）" + + +def get_result(workspace_dir: str, sid: str) -> dict: + sdir = sub_dir(workspace_dir, sid) + meta = _stale_guard(_read_meta(sdir)) + rpath = os.path.join(sdir, "result.txt") + text = "" + if os.path.exists(rpath): + with open(rpath, "r", encoding="utf-8", errors="replace") as f: + text = f.read()[-RESULT_LIMIT:] + return {"meta": meta, "result": text or "(暂无输出)"}