feat(agent): 会话agent四项Hermes能力对齐(记忆写入/技能管理/后台进程/并行委派)

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

@@ -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"
+
     # ═══════════════════════════════════════════════════════
     # 上下文压缩
     # ═══════════════════════════════════════════════════════

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": "已发送终止信号"}

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, '未实现的动作分支', {}

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 "(暂无输出)"}