feat(agent): 记忆多租户隔离+memory/manage_skill/process/subagent工具对齐Hermes

- memory_store: org_id/user_id归属列、可见性规则(visible_to)、key掺租户维度、
  分区淘汰(不碰种子/他租户)、写后失效缓存(多worker一致)
- agent_config: GENERAL_TOOLS 新增 memory/manage_skill/process/subagent 四工具;
  run_command 加 background/timeout; ToolDefinition 加显式 required 字段
  (可选参数不再被 native FC 标成 required)
- tool_registry: to_openai_schema 尊重显式 required
- models: pipeline_user_memory 加 org_id/user_id(配套迁移 m0023)

models/pipeline_user_memory.json

@@ -54,6 +54,22 @@
       "nullable": "no",
       "default": ""
     },
+    {
+      "name": "org_id",
+      "title": "归属机构ID(空=平台种子)",
+      "type": "str",
+      "length": 32,
+      "nullable": "no",
+      "default": ""
+    },
+    {
+      "name": "user_id",
+      "title": "归属用户ID(user域必填)",
+      "type": "str",
+      "length": 32,
+      "nullable": "no",
+      "default": ""
+    },
     {
       "name": "priority",
       "title": "优先级",

pipeline_core/agent_config.py

@@ -27,6 +27,10 @@ class ToolDefinition:
     enabled: bool = True
     category: str = "general"  # project / task / repo / shell / agent
     requires_confirmation: bool = False  # 是否需要用户确认
+    # 显式必填参数列表（2026-09-10）：None = 旧行为（全部参数必填）。
+    # 含可选参数的工具必须显式声明，否则 native FC schema 把可选参数
+    # 也标成 required，LLM 被迫为每个参数编值（v1 同款坑，机制层根治）。
+    required: Optional[list] = None
 
 
 @dataclass
@@ -132,6 +136,8 @@ class AgentConfig:
                     "parameters": t.parameters,
                     "enabled": t.enabled,
                     "category": t.category,
+                    "requires_confirmation": t.requires_confirmation,
+                    "required": t.required,
                 }
                 for t in self.tools
             ],
@@ -241,8 +247,9 @@ GENERAL_TOOLS = [
     ),
     ToolDefinition(
         name="run_command",
-        description="在工作空间中执行shell命令",
+        description="在工作空间中执行shell命令。长任务（装依赖/构建/跑测试）用 background=true 转后台：立即返回 job_id，之后用 process 工具 poll/log/wait/kill 跟进，不阻塞对话",
-        parameters={"command": "命令"},
+        parameters={"command": "命令", "background": "可选：true=后台执行返回job_id（默认前台，60秒超时）", "timeout": "可选：前台超时秒数（默认60，最大300）"},
+        required=["command"],
         category="shell",
         requires_confirmation=True,
     ),
@@ -255,8 +262,16 @@ GENERAL_TOOLS = [
     ),
     ToolDefinition(
         name="delegate_subtask",
-        description="派生子agent调查子任务（并行执行）",
+        description="派生子agent执行子任务。background=false（默认）同步等待返回结果；background=true 后台并行执行（同一会话最多3个并行），立即返回 subagent_id，之后用 subagent 工具跟进。子agent看不到本会话上下文，goal/context 必须自包含",
-        parameters={"goal": "子任务目标", "context": "背景信息"},
+        parameters={"goal": "子任务目标（自包含）", "context": "可选：背景信息", "background": "可选：true=后台并行（默认false同步等待）"},
+        required=["goal"],
+        category="agent",
+    ),
+    ToolDefinition(
+        name="subagent",
+        description="管理后台子agent：list=列出本会话的子agent及状态；steer=给运行中的子agent追加指示（下轮生效）；stop=提前终止（返回已有部分结果）；result=取最终结果",
+        parameters={"action": "list|steer|stop|result", "subagent_id": "子agent ID（list 时可空）", "message": "steer 时的追加指示"},
+        required=["action"],
         category="agent",
     ),
     # ── 通用工具集（Hermes CLI 能力子集：文件/搜索/会话/规划）──
@@ -264,12 +279,14 @@ GENERAL_TOOLS = [
         name="read_file",
         description="读取工作空间中的文件内容。支持 docx/pdf/txt/md/json 等格式，docx/pdf 会自动解析提取正文文本，直接调用即可读取。单次最多 30000 字符；返回带截断提示时，用 offset 参数分段续读后文，逐段读完全文（切勿只读开头就以为读全了）",
         parameters={"path": "相对路径", "offset": "可选：从第几个字符开始读（分段读大文件）"},
+        required=["path"],
         category="file",
     ),
     ToolDefinition(
         name="load_skill",
         description="加载指定技能的完整内容（具体步骤/规范/陷阱）。先在系统提示的『可用技能』目录里找到技能名，需要时再调用本工具加载正文。技能带子文件（模板库/参考文档）时，正文末尾会列出可用子文件，用 file_path 参数逐个加载",
         parameters={"name": "技能名称", "file_path": "可选：子文件相对路径（如 references/uapi-templates.md），只允许 references/scripts/templates/assets 下的文件"},
+        required=["name"],
         category="skill",
     ),
     ToolDefinition(
@@ -288,6 +305,7 @@ GENERAL_TOOLS = [
         name="propose_skill",
         description="沉淀技能：把值得复用的流程/经验/坑写成技能并实时发布到你所属机构的技能目录（同名覆盖通用技能，机构内立即生效，其他机构不受影响）。用户要求沉淀经验、总结技能，或你发现反复出现的流程/坑/规范时调用。平台缺省机构(org 0)的提议会转人工审核（不实时生效）",
         parameters={"name": "技能名（字母数字._-，≤64字符）", "description": "技能描述", "content": "SKILL.md 正文（frontmatter 可选，系统自动补全）"},
+        required=["name", "content"],
         category="skill",
     ),
     ToolDefinition(
@@ -300,12 +318,14 @@ GENERAL_TOOLS = [
         name="list_files",
         description="列出工作空间目录内容",
         parameters={"path": "相对路径(可选,默认工作空间根)"},
+        required=[],
         category="file",
     ),
     ToolDefinition(
         name="search_files",
         description="在工作空间中搜索文件内容（grep）",
         parameters={"pattern": "搜索关键词或正则", "path": "相对路径(可选,默认整个工作空间)"},
+        required=["pattern"],
         category="file",
     ),
     ToolDefinition(
@@ -314,10 +334,56 @@ GENERAL_TOOLS = [
         parameters={"query": "搜索关键词"},
         category="memory",
     ),
+    # ── 持久记忆写入（2026-09-10，对齐 Hermes memory 工具）──
+    # 多租户门禁在 agent_loop_v2._t_memory：agent 只能写自己机构/自己名下的
+    # user/project/pipeline 域，global/org 平台种子域禁写；org_id 强制注入。
+    ToolDefinition(
+        name="memory",
+        description="持久记忆（跨会话保留）：add=记住一条事实/偏好；list=列出你可见的记忆；remove=删除一条。用户表达偏好/纠正/要求记住某事，或你发现值得跨会话保留的事实时调用。scope 选择：user=关于用户本人的偏好，project=当前项目的约定/事实，pipeline=当前产线的通用经验",
+        parameters={
+            "action": "add|list|remove",
+            "content": "记忆内容（add 必填，一句话陈述事实，勿写指令式）",
+            "scope": "可选：user|project|pipeline（默认 user；无项目时 project 不可用）",
+            "category": "可选：user=用户偏好|memory=一般事实（默认 memory）",
+            "key": "remove 时必填：记忆键（list 返回）",
+        },
+        required=["action"],
+        category="memory",
+    ),
+    # ── 技能管理（2026-09-10，对齐 Hermes skill_manage：create/patch/write_file/remove_file/delete）──
+    # 隔离由 skill_live.resolve_target 保证：只能落在本机构 orgs/{org}/ 或
+    # org 0 降级 users/{uid}/，global 与他机构目录物理不可达。
+    ToolDefinition(
+        name="manage_skill",
+        description="管理你所属机构的技能库：create=新建技能（整篇 SKILL.md）；patch=定向修改已有技能的片段（old_string 须唯一）；write_file=给技能添加子文件（限 references/scripts/templates/assets 下）；remove_file=删技能子文件；delete=删除整个技能。只能操作本机构（或本人）目录下的技能——覆盖自通用技能的同名副本可改可删，但删不掉通用(global)原版。propose_skill 等价于 create",
+        parameters={
+            "action": "create|patch|write_file|remove_file|delete",
+            "name": "技能名（字母数字._-，≤64字符）",
+            "description": "create 时的一句话描述",
+            "content": "create 时的 SKILL.md 正文；patch 时不用",
+            "old_string": "patch 必填：要替换的原文片段（须在目标文件中唯一）",
+            "new_string": "patch 必填：替换后的新文本",
+            "file_path": "write_file/remove_file 必填、patch 可选：子文件相对路径（如 references/api.md，首段限 references/scripts/templates/assets）",
+            "file_content": "write_file 必填：子文件内容",
+        },
+        required=["action", "name"],
+        category="skill",
+    ),
+    # ── 后台进程管理（2026-09-10，配套 run_command background=true）──
+    # 状态文件化（workspace/.bg/{job_id}/），跨 worker 进程可读；
+    # 隔离靠 workspace 路径（用户/项目工作空间天然隔离）。
+    ToolDefinition(
+        name="process",
+        description="管理 run_command 启动的后台任务：poll=查状态+新增输出；log=取完整输出（可分页）；wait=阻塞等待结束（最多120秒，超时返回部分输出）；kill=终止",
+        parameters={"action": "poll|log|wait|kill", "job_id": "后台任务 ID（run_command background=true 返回）", "offset": "log 可选：从第几个字符开始读"},
+        required=["action", "job_id"],
+        category="shell",
+    ),
     ToolDefinition(
         name="todo",
         description="管理当前会话的任务清单（list/add/done）",
         parameters={"action": "list|add|done", "content": "任务内容(add时必填)"},
+        required=["action"],
         category="agent",
     ),
     # ── 平台模型调用（2026-09-07）：可用模型 = 平台 owner 机构 + 本机构的
@@ -330,6 +396,7 @@ GENERAL_TOOLS = [
         name="list_platform_models",
         description="列出平台当前可用的模型（本机构+平台owner机构的模型，含能力类型：t2t对话/i2t图像理解/t2i文生图/t2v文生视频/i2v图生视频/tts语音合成/asr语音识别等）。用户问「有哪些模型/能做什么」，或你要调用非对话能力（生图/视频/语音）前需要选模型时调用。capability 参数可按能力过滤（如 t2i）",
         parameters={"capability": "可选：能力类型过滤（如 t2i/t2v/tts），空=全部"},
+        required=[],
         category="model",
     ),
     ToolDefinition(
@@ -341,6 +408,7 @@ GENERAL_TOOLS = [
             "capability": "可选：能力类型（t2i/t2v/i2v/tts/asr等，自动匹配时用于过滤候选）",
             "params": "可选：业务参数 JSON 字符串。媒体输入用三数组契约：image_files/audio_files/video_files（值为公网URL或base64的数组）；生成参数如 resolution/duration/size 按模型文档",
         },
+        required=["task"],
         category="model",
     ),
     # ── 联网检索与网页抓取（2026-09-08，对齐 Hermes web_search/web_extract）──
@@ -350,6 +418,7 @@ GENERAL_TOOLS = [
         name="web_search",
         description="联网检索信息（搜索引擎）。需要外部资料/时事/文档/依据而知识库与本地文件没有时调用。返回标题+URL+摘要列表；需要某条结果的完整内容时再用 fetch_url 抓取",
         parameters={"query": "搜索关键词", "limit": "可选：返回条数(默认8,最大10)"},
+        required=["query"],
         category="web",
     ),
     ToolDefinition(
@@ -370,6 +439,7 @@ GENERAL_TOOLS = [
             "order_by": "可选：排序列（可带 ASC/DESC）",
             "limit": "可选：行数上限（默认/最大100）",
         },
+        required=["table"],
         category="data",
     ),
 ]

pipeline_core/memory_store.py

@@ -1,5 +1,5 @@
-"""
+# -*- coding: utf-8 -*-
-pipeline-core: Memory Store — 跨会话记忆系统
+"""pipeline-core: Memory Store — 跨会话记忆系统（多租户版，2026-09-10）
 
 对照 Hermes Agent 的 memory 工具 + MEMORY.md / USER.md：
 - 用户偏好存储（user profile）
@@ -8,25 +8,44 @@ pipeline-core: Memory Store — 跨会话记忆系统
 - 自动去重和淘汰
 
 存储方式：MySQL pipeline_user_memory 表 + 内存缓存
+
+多租户隔离（2026-09-10，服务多机构多用户）：
+- 每条记忆带 org_id / user_id 归属列（org_id='' 且 user_id='' = 平台种子记忆）
+- memory_key = hash(content + org_id + user_id)——同内容不同租户 key 不同，
+  唯一键永不跨租户碰撞
+- 可见性规则（visible_to）：
+    global 种子（org_id=''）           → 所有人可见（平台预置，agent 禁写）
+    org 归属（org_id=O）               → 仅机构 O 的用户可见
+    user 归属（user_id=U）             → 仅用户 U 本人可见
+    pipeline/project 域带 org_id=O     → 机构 O 内、且产线/项目匹配才可见
+- 写入约束由调用方（agent handler）把关：agent 只能写自己 org_id 名下、
+  scope ∈ {user, project, pipeline}；global/org 域是平台种子域，禁写
+- 淘汰分区（_evict）：只淘汰「同 org_id + 同 user 分区」内的低优先级旧条目，
+  绝不删平台种子（org_id=''）或其他机构的记忆
 """
 
-import json
+import hashlib
 import logging
 import time
-from dataclasses import dataclass, field
+from dataclasses import dataclass
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Tuple
 
 logger = logging.getLogger("pipeline.memory_store")
 
+# 缓存分区键 = (category, scope, scope_id, org_id, user_id)
+_PartitionKey = Tuple[str, str, str, str, str]
+
 
 @dataclass
 class MemoryEntry:
     """一条记忆"""
-    key: str                         # 唯一标识
+    key: str                         # 唯一标识（掺入租户维度的内容哈希）
     content: str                     # 记忆内容
     category: str = "memory"         # "user" | "memory"
-    scope: str = "global"            # "global" | "pipeline" | "project" | "user"（归属范围）
+    scope: str = "global"            # "global" | "pipeline" | "project" | "org" | "user"
-    scope_id: str = ""               # pipeline_id / project_id / user_id
+    scope_id: str = ""               # pipeline_id / project_id / org_id / user_id
+    org_id: str = ""                 # 归属机构（'' = 平台种子）
+    user_id: str = ""                # 归属用户（user 域必填；'' = 非个人）
     priority: int = 0                # 优先级（越高越重要）
     created_at: float = 0.0
     updated_at: float = 0.0
@@ -39,6 +58,8 @@ class MemoryEntry:
             "category": self.category,
             "scope": self.scope,
             "scope_id": self.scope_id,
+            "org_id": self.org_id,
+            "user_id": self.user_id,
             "priority": self.priority,
             "created_at": self.created_at,
             "updated_at": self.updated_at,
@@ -53,19 +74,38 @@ class MemoryEntry:
             category=data.get("category", "memory"),
             scope=data.get("scope", "global"),
             scope_id=data.get("scope_id", ""),
+            org_id=data.get("org_id", ""),
+            user_id=data.get("user_id", ""),
             priority=data.get("priority", 0),
             created_at=data.get("created_at", time.time()),
             updated_at=data.get("updated_at", time.time()),
             access_count=data.get("access_count", 0),
         )
 
+    @property
+    def partition(self) -> _PartitionKey:
+        return (self.category, self.scope, self.scope_id, self.org_id, self.user_id)
+
+
+def _make_key(content: str, org_id: str = "", user_id: str = "",
+              scope: str = "", scope_id: str = "") -> str:
+    """从内容 + 租户 + 作用域维度生成稳定 key。
+
+    掺入 org_id/user_id/scope/scope_id：
+    - 同一段内容在机构 A 和机构 B 各自存储时 key 不同，唯一索引
+      (memory_key, category) 永不跨租户碰撞、互不覆盖；
+    - 同租户同内容写到不同作用域（项目域/个人域）也是不同条目。
+    """
+    raw = f"{content}|org={org_id}|user={user_id}|scope={scope}|sid={scope_id}"
+    return hashlib.md5(raw.encode("utf-8")).hexdigest()[:16]
+
 
 class MemoryStore:
-    """跨会话记忆存储。
+    """跨会话记忆存储（多租户）。
 
     两层架构：
     1. MySQL pipeline_user_memory（持久化）
-    2. 内存缓存（加速读取）
+    2. 内存缓存（按分区键组织，加速读取）
     """
 
     # 优先级分组
@@ -73,12 +113,18 @@ class MemoryStore:
     PRIORITY_MEDIUM = 5      # 项目约定、工作流
     PRIORITY_LOW = 1         # 临时备注
 
+    # agent 可写的 scope 白名单（global/org 是平台种子域，禁写）
+    WRITABLE_SCOPES = ("user", "project", "pipeline")
+
     def __init__(self):
-        self._cache: Dict[str, Dict[str, MemoryEntry]] = {}  # {category: {key: entry}}
+        # {partition_key: {key: MemoryEntry}}
+        self._cache: Dict[_PartitionKey, Dict[str, MemoryEntry]] = {}
         self._cache_loaded = False
 
+    # ── 缓存 ──
+
     async def _ensure_cache(self):
-        """从 DB 加载缓存"""
+        """从 DB 加载缓存（全表按优先级取前 2000 条；可见性在读取时过滤）"""
         if self._cache_loaded:
             return
 
@@ -88,8 +134,9 @@ class MemoryStore:
             db = DBPools()
             async with db.sqlorContext("pipeline") as sor:
                 recs = await sor.sqlExe(
-                    "SELECT memory_key, content, category, scope, scope_id, priority, created_at, updated_at, access_count "
+                    "SELECT memory_key, content, category, scope, scope_id, "
-                    "FROM pipeline_user_memory ORDER BY priority DESC, updated_at DESC LIMIT 500",
+                    "org_id, user_id, priority, created_at, updated_at, access_count "
+                    "FROM pipeline_user_memory ORDER BY priority DESC, updated_at DESC LIMIT 2000",
                     {},
                 )
                 for r in (recs or []):
@@ -99,30 +146,82 @@ class MemoryStore:
                         category=getattr(r, "category", "memory"),
                         scope=getattr(r, "scope", "global"),
                         scope_id=getattr(r, "scope_id", ""),
+                        org_id=getattr(r, "org_id", "") or "",
+                        user_id=getattr(r, "user_id", "") or "",
                         priority=getattr(r, "priority", 0),
                         created_at=_ts(getattr(r, "created_at", None)),
                         updated_at=_ts(getattr(r, "updated_at", None)),
                         access_count=getattr(r, "access_count", 0),
                     )
-                    if entry.category not in self._cache:
+                    pk = entry.partition
-                        self._cache[entry.category] = {}
+                    if pk not in self._cache:
-                    self._cache[entry.category][entry.key] = entry
+                        self._cache[pk] = {}
+                    self._cache[pk][entry.key] = entry
         except Exception as e:
             logger.warning(f"MemoryStore cache load failed: {e}")
 
         self._cache_loaded = True
 
+    def _invalidate_cache(self):
+        """多 worker 共享 DB：写路径失效缓存，下次读重新加载（防跨进程读到旧缓存）。"""
+        self._cache = {}
+        self._cache_loaded = False
+
+    # ── 可见性 ──
+
+    @staticmethod
+    def visible_to(entry: MemoryEntry, org_id: str, user_id: str,
+                   pipeline_id: str = "", project_id: str = "") -> bool:
+        """多租户可见性判定（唯一权威规则，读取端全部走这里）。
+
+        - 平台种子（org_id='' 且 user_id=''）→ 所有人可见（只读预置）
+        - user 归属 → 仅本人
+        - org 归属（scope=org）→ 仅本机构
+        - pipeline 域 → 本机构（或种子）且 pipeline_id 匹配
+        - project 域 → 本机构（或种子）且 project_id 匹配
+        - global 域机构记忆（org_id=O）→ 仅机构 O
+        """
+        seed = (entry.org_id == "" and entry.user_id == "")
+        if seed:
+            # 种子记忆：pipeline/project 域还要求当前上下文匹配（或域值为空=通用种子）
+            if entry.scope == "pipeline" and entry.scope_id and entry.scope_id != pipeline_id:
+                return False
+            if entry.scope == "project" and entry.scope_id and entry.scope_id != project_id:
+                return False
+            return True
+        # 归属记忆：机构必须匹配（跨机构一律不可见）
+        if entry.org_id and entry.org_id != org_id:
+            return False
+        # user 域：仅本人可见（user_id 记录时恒为归属者）
+        if entry.scope == "user":
+            return entry.user_id == user_id
+        # project/pipeline 域：机构内共享（user_id 只记创建者，供删除过滤，
+        # 不影响可见性）；上下文须匹配对应项目/产线
+        if entry.scope == "pipeline":
+            return entry.scope_id == pipeline_id
+        if entry.scope == "project":
+            return entry.scope_id == project_id
+        if entry.scope == "org":
+            return True   # org_id 已在上面匹配
+        # 其余（global 域带 org_id 的机构通用记忆）：机构匹配即可见
+        return True
+
+    # ── 写入 ──
+
     async def add(self, content: str, category: str = "memory",
                   priority: int = PRIORITY_MEDIUM, key: str = None,
-                  scope: str = "global", scope_id: str = ""):
+                  scope: str = "global", scope_id: str = "",
-        """添加一条记忆。自动生成 key（取内容前80字符的哈希）。
+                  org_id: str = "", user_id: str = ""):
+        """添加一条记忆（自动生成掺租户维度的 key）。
 
-        scope 分域：global（通用）/ pipeline（产线）/ project（项目）/ user（用户个人）。
+        ⚠️ 本方法不做写入权限校验（平台内部/种子导入也用它）。
+        agent 侧的写入门禁（禁 global/org 域、org_id 强制注入）在
+        agent_loop_v2 的 _t_memory handler 里把关，勿绕过 handler 直调。
         """
         await self._ensure_cache()
 
         if not key:
-            key = _make_key(content)
+            key = _make_key(content, org_id, user_id, scope, scope_id)
 
         now = time.time()
         entry = MemoryEntry(
@@ -131,87 +230,154 @@ class MemoryStore:
             category=category,
             scope=scope,
             scope_id=scope_id,
+            org_id=org_id,
+            user_id=user_id,
             priority=priority,
             created_at=now,
             updated_at=now,
         )
 
-        # 更新缓存
+        pk = entry.partition
-        if category not in self._cache:
+        if pk not in self._cache:
-            self._cache[category] = {}
+            self._cache[pk] = {}
-        self._cache[category][key] = entry
+        self._cache[pk][key] = entry
 
-        # 持久化
         try:
             await self._db_upsert(entry)
         except Exception as e:
             logger.error(f"MemoryStore add failed: {e}")
 
-        # 淘汰低优先级旧条目
+        # 分区内淘汰（只动同 org+user 分区，绝不碰种子和他租户）
-        await self._evict(max_entries=100)
+        await self._evict_partition(org_id, user_id, max_entries=100)
 
-    async def get(self, category: str = None, key: str = None,
+        # 多 worker 一致性：写后失效本进程缓存，下次读取重载全表——
-                  scope: str = None, scope_id: str = None) -> List[MemoryEntry]:
+        # 否则其他 worker 进程写入的记忆本进程永远看不到（旧版同款缺陷）
-        """获取记忆。可按分类/key/scope 筛选。
+        self._invalidate_cache()
 
-        scope 过滤：scope=None 返回全部；scope 指定时返回该 scope 且 scope_id 匹配（或 scope_id=None 时该 scope 全部）。
+    # ── 读取 ──
+
+    async def get(self, category: Optional[str] = None, key: Optional[str] = None,
+                  scope: Optional[str] = None, scope_id: Optional[str] = None,
+                  org_id: Optional[str] = None, user_id: Optional[str] = None) -> List[MemoryEntry]:
+        """获取记忆（**不带可见性过滤的底层查询**，管理/巡检用）。
+
+        agent 读取一律走 get_visible()——本方法会把所有租户的条目都返回，
+        直接暴露给 LLM 就是跨机构泄漏。
         """
         await self._ensure_cache()
 
         results = []
-        cats = [category] if category else list(self._cache.keys())
+        for pk, entries in self._cache.items():
-        for cat in cats:
+            cat, sc, sid, oid, uid = pk
-            if cat in self._cache:
+            if category and cat != category:
-                for k, entry in self._cache[cat].items():
+                continue
-                    if key and k != key:
+            if scope is not None and sc != scope:
-                        continue
+                continue
-                    if scope is not None:
+            if scope_id is not None and sid != scope_id:
-                        if entry.scope != scope:
+                continue
-                            continue
+            if org_id is not None and oid != org_id:
-                        if scope_id is not None and entry.scope_id != scope_id:
+                continue
-                            continue
+            if user_id is not None and uid != user_id:
-                    results.append(entry)
+                continue
-                    entry.access_count += 1
+            for k, entry in entries.items():
+                if key and k != key:
+                    continue
+                results.append(entry)
+                entry.access_count += 1
 
         results.sort(key=lambda e: e.priority, reverse=True)
         return results
 
-    async def remove(self, key: str, category: str = "memory"):
+    async def get_visible(self, org_id: str, user_id: str,
-        """删除一条记忆"""
+                          pipeline_id: str = "", project_id: str = "",
+                          category: str = None) -> List[MemoryEntry]:
+        """按租户可见性取记忆（agent 读取唯一入口）。"""
+        all_entries = await self.get(category=category)
+        visible = [e for e in all_entries
+                   if self.visible_to(e, org_id or "", user_id or "",
+                                      pipeline_id or "", project_id or "")]
+        visible.sort(key=lambda e: e.priority, reverse=True)
+        return visible
+
+    # ── 删除 ──
+
+    async def remove(self, key: str, category: str = "memory",
+                     scope: Optional[str] = None, scope_id: str = "",
+                     org_id: Optional[str] = None, user_id: Optional[str] = None):
+        """删除一条记忆。
+
+        多租户安全：org_id/user_id 传 None 表示不过滤（内部淘汰用，已分区）；
+        agent 侧删除必须显式传 org_id/user_id，防止按 key 误删他租户同名条目。
+        """
         await self._ensure_cache()
-        if category in self._cache:
+        removed = 0
-            self._cache[category].pop(key, None)
+        for pk in list(self._cache.keys()):
+            cat, sc, sid, oid, uid = pk
+            if cat != category:
+                continue
+            if scope is not None and sc != scope:
+                continue
+            if scope_id and sid != scope_id:
+                continue
+            if org_id is not None and oid != org_id:
+                continue
+            if user_id is not None and uid != user_id:
+                continue
+            if key in self._cache[pk]:
+                self._cache[pk].pop(key, None)
+                removed += 1
 
-        try:
+        if removed:
-            from sqlor.dbpools import DBPools
+            try:
+                from sqlor.dbpools import DBPools
 
-            db = DBPools()
+                db = DBPools()
-            async with db.sqlorContext("pipeline") as sor:
+                async with db.sqlorContext("pipeline") as sor:
-                await sor.sqlExe(
+                    sql = "DELETE FROM pipeline_user_memory WHERE memory_key=${k}$ AND category=${c}$"
-                    "DELETE FROM pipeline_user_memory WHERE memory_key=${k}$ AND category=${c}$",
+                    params = {"k": key, "c": category}
-                    {"k": key, "c": category},
+                    if scope is not None:
-                )
+                        sql += " AND scope=${s}$"
-        except Exception as e:
+                        params["s"] = scope
-            logger.error(f"MemoryStore remove failed: {e}")
+                    if scope_id:
+                        sql += " AND scope_id=${sid}$"
+                        params["sid"] = scope_id
+                    if org_id is not None:
+                        sql += " AND org_id=${oid}$"
+                        params["oid"] = org_id
+                    if user_id is not None:
+                        sql += " AND user_id=${uid}$"
+                        params["uid"] = user_id
+                    await sor.sqlExe(sql, params)
+                self._invalidate_cache()   # 同 add：写后失效，跨进程可见
+            except Exception as e:
+                logger.error(f"MemoryStore remove failed: {e}")
+        return removed
 
-    async def build_prompt_block(self, category: str = None, max_entries: int = 20,
+    # ── prompt 注入 ──
-                                 scope: str = None, scope_id: str = None) -> str:
+
-        """构建注入 system prompt 的记忆段落。
+    async def build_prompt_block(self, category: Optional[str] = None, max_entries: int = 20,
+                                 scope: Optional[str] = None, scope_id: Optional[str] = None,
+                                 org_id: str = "", user_id: str = "",
+                                 pipeline_id: str = "", project_id: str = "") -> str:
+        """构建注入 system prompt 的记忆段落（租户可见性过滤版）。
+
+        传入 org_id/user_id 时按 visible_to 过滤（agent 会话注入的唯一正确用法）；
+        两者都不传 = 旧行为（global 种子 + 指定 scope），仅供无租户上下文的
+        内部场景（如平台巡检），agent 链路禁用。
 
-        指定 scope 时加载「global（通用）+ 指定 scope（专属）」的记忆，
-        实现通用记忆 + 产线/项目专属记忆的叠加注入。
         高优先级记忆注入完整内容，低优先级只注入摘要。
         """
-        if scope:
+        if org_id or user_id:
-            # 通用 + 专属叠加
+            entries = await self.get_visible(org_id, user_id, pipeline_id, project_id,
-            global_entries = await self.get(category, scope="global")
+                                             category=category)
-            scoped_entries = await self.get(category, scope=scope, scope_id=scope_id)
-            # 去重（同 key 专属覆盖通用）
-            merged = {e.key: e for e in global_entries}
-            merged.update({e.key: e for e in scoped_entries})
-            entries = list(merged.values())
         else:
-            entries = await self.get(category)
+            # 兼容旧签名：无租户上下文时只给平台种子（绝不给归属记忆）
+            all_entries = await self.get(category=category)
+            entries = [e for e in all_entries if e.org_id == "" and e.user_id == ""]
+            if scope:
+                entries = [e for e in entries
+                           if e.scope in ("global", scope)
+                           and (not scope_id or e.scope != scope or e.scope_id == scope_id)]
 
         if not entries:
             return ""
@@ -231,16 +397,19 @@ class MemoryStore:
 
         return "\n".join(blocks) if blocks else ""
 
+    # ── DB ──
+
     async def _db_upsert(self, entry: MemoryEntry):
-        """写入/更新 DB（同 key + category + scope + scope_id 才 upsert）。"""
+        """写入/更新 DB（同 key + category + scope + scope_id + org + user 才 upsert）。"""
         from sqlor.dbpools import DBPools
 
         db = DBPools()
         async with db.sqlorContext("pipeline") as sor:
             existing = await sor.sqlExe(
                 "SELECT id FROM pipeline_user_memory WHERE memory_key=${k}$ AND category=${c}$ "
-                "AND scope=${s}$ AND scope_id=${sid}$",
+                "AND scope=${s}$ AND scope_id=${sid}$ AND org_id=${oid}$ AND user_id=${uid}$",
-                {"k": entry.key, "c": entry.category, "s": entry.scope, "sid": entry.scope_id},
+                {"k": entry.key, "c": entry.category, "s": entry.scope, "sid": entry.scope_id,
+                 "oid": entry.org_id, "uid": entry.user_id},
             )
             if existing:
                 rid = existing[0].id
@@ -261,36 +430,40 @@ class MemoryStore:
                     "category": entry.category,
                     "scope": entry.scope,
                     "scope_id": entry.scope_id,
+                    "org_id": entry.org_id,
+                    "user_id": entry.user_id,
                     "priority": entry.priority,
                 })
 
-    async def _evict(self, max_entries: int = 100):
+    async def _evict_partition(self, org_id: str, user_id: str, max_entries: int = 100):
-        """淘汰低优先级的旧记忆（LRU）"""
+        """分区淘汰：只淘汰「同 org_id + 同 user_id」分区内的低优先级旧记忆。
-        total = sum(len(v) for v in self._cache.values())
+
-        if total <= max_entries:
+        平台种子（org_id='' 且 user_id=''）永不被 agent 写入触发淘汰；
+        机构 A 的写入永远不会删机构 B / 其他用户 / 种子的记忆。
+        """
+        partition_entries: List[MemoryEntry] = []
+        for pk, entries in self._cache.items():
+            _cat, _sc, _sid, oid, uid = pk
+            if oid == (org_id or "") and uid == (user_id or ""):
+                partition_entries.extend(entries.values())
+
+        if len(partition_entries) <= max_entries:
             return
 
-        # 找出可淘汰的条目（低优先级 + 低访问次数）
+        partition_entries.sort(key=lambda e: (e.priority, e.access_count, e.updated_at))
-        all_entries = []
+        to_remove = partition_entries[: (len(partition_entries) - max_entries)]
-        for cat_entries in self._cache.values():
-            all_entries.extend(cat_entries.values())
-
-        all_entries.sort(key=lambda e: (e.priority, e.access_count, e.updated_at))
-
-        to_remove = all_entries[: (total - max_entries)]
         for entry in to_remove:
-            await self.remove(entry.key, entry.category)
+            await self.remove(entry.key, entry.category,
+                              scope=entry.scope, scope_id=entry.scope_id,
+                              org_id=entry.org_id, user_id=entry.user_id)
+
+    # 兼容旧接口名（旧调用点 _evict(max_entries) 语义已不安全，转发到分区版）
+    async def _evict(self, max_entries: int = 100, org_id: str = "", user_id: str = ""):
+        await self._evict_partition(org_id, user_id, max_entries)
 
 
 # ── 辅助 ──
 
-def _make_key(content: str) -> str:
-    """从内容生成稳定 key"""
-    import hashlib
-
-    return hashlib.md5(content.encode("utf-8")).hexdigest()[:16]
-
-
 def _ts(val) -> float:
     """转换时间戳"""
     if val is None:

pipeline_core/tool_registry.py

@@ -88,7 +88,8 @@ class ToolRegistry:
                             k: {"type": "string", "description": v}
                             for k, v in (t.parameters or {}).items()
                         },
-                        "required": list(t.parameters.keys()) if t.parameters else [],
+                        "required": (t.required if t.required is not None
+                                     else list(t.parameters.keys())) if t.parameters else [],
                     },
                 },
             }