feat: 角色定义可插拔(RoleSpec+辅助函数)+记忆分域(scope:global/pipeline/project/user)

pipeline_core/__init__.py

@@ -30,11 +30,16 @@ from .agent_config import (
 )
 from .ability import (
     PipelineAbility,
+    RoleSpec,
     register_ability,
     get_ability,
     list_abilities,
     get_ability_tools,
     get_ability_handler,
+    get_role_spec,
+    normalize_role,
+    get_next_role,
+    list_roles,
 )
 from .slash import (
     SlashCommand,

pipeline_core/ability.py

@@ -19,9 +19,25 @@ from typing import Callable, Dict, List, Optional
 logger = logging.getLogger("pipeline.ability")
 
 
+@dataclass
+class RoleSpec:
+    """产线内角色定义（可插拔）。
+
+    一个角色 = 工具白名单 + 专属 prompt + 专属模型 + 任务链位置 + 别名。
+    产线能力包通过 roles 字段定义自己的角色集（如 SDLC 的 requirement/design/develop/test/deploy）。
+    """
+    name: str                                    # "develop"
+    description: str = ""                        # 角色职责描述
+    aliases: List[str] = field(default_factory=list)   # ["dev", "developer"]
+    system_prompt: str = ""                      # 角色专属 prompt（职责 + 产出要求）
+    next_role: str = ""                          # 任务链下一角色（空 = 终结）
+    model_name: str = ""                         # 角色专属模型（空 = 继承产线）
+    tools: List[str] = field(default_factory=list)      # 工具白名单（空 = 产线全部工具）
+
+
 @dataclass
 class PipelineAbility:
-    """产线能力包：一个产线 = 一组工具 + 一段 prompt + 一组 handler。
+    """产线能力包：一个产线 = 一组工具 + 一段 prompt + 一组 handler + 一组角色。
 
     handlers 签名统一为：async def handler(sor, params, ctx) -> str
       - sor: 已打开的 pipeline DB context（由 AgentExecutor 统一管理）
@@ -33,6 +49,7 @@ class PipelineAbility:
     tools: List = field(default_factory=list)           # [ToolDefinition]
     system_prompt: str = ""                             # 产线专属 prompt 片段（追加到通用心智后）
     handlers: Dict[str, Callable] = field(default_factory=dict)  # {tool_name: handler}
+    roles: List[RoleSpec] = field(default_factory=list)         # 产线角色集
 
 
 # ── 全局注册表 ──
@@ -71,3 +88,37 @@ def get_ability_handler(pipeline_id: str, tool_name: str) -> Optional[Callable]:
     if not a:
         return None
     return a.handlers.get(tool_name)
+
+
+# ── 角色查询 ──
+
+def get_role_spec(pipeline_id: str, role: str) -> Optional[RoleSpec]:
+    """取某产线某角色的定义（按 name 或 alias 匹配）。"""
+    a = get_ability(pipeline_id)
+    if not a or not role:
+        return None
+    r = (role or "").strip().lower()
+    for spec in a.roles:
+        if spec.name.lower() == r or r in [x.lower() for x in spec.aliases]:
+            return spec
+    return None
+
+
+def normalize_role(pipeline_id: str, role: str) -> str:
+    """角色标准化：别名 → 规范名。能力包未定义时原样返回（lower）。"""
+    spec = get_role_spec(pipeline_id, role)
+    if spec:
+        return spec.name
+    return (role or "").strip().lower()
+
+
+def get_next_role(pipeline_id: str, role: str) -> str:
+    """任务链下一角色（空 = 终结）。"""
+    spec = get_role_spec(pipeline_id, role)
+    return spec.next_role if spec else ""
+
+
+def list_roles(pipeline_id: str) -> List[RoleSpec]:
+    """列出某产线的全部角色。"""
+    a = get_ability(pipeline_id)
+    return a.roles if a else []

pipeline_core/memory_store.py

@@ -25,6 +25,8 @@ class MemoryEntry:
     key: str                         # 唯一标识
     content: str                     # 记忆内容
     category: str = "memory"         # "user" | "memory"
+    scope: str = "global"            # "global" | "pipeline" | "project" | "user"（归属范围）
+    scope_id: str = ""               # pipeline_id / project_id / user_id
     priority: int = 0                # 优先级（越高越重要）
     created_at: float = 0.0
     updated_at: float = 0.0
@@ -35,6 +37,8 @@ class MemoryEntry:
             "key": self.key,
             "content": self.content,
             "category": self.category,
+            "scope": self.scope,
+            "scope_id": self.scope_id,
             "priority": self.priority,
             "created_at": self.created_at,
             "updated_at": self.updated_at,
@@ -47,6 +51,8 @@ class MemoryEntry:
             key=data.get("key", ""),
             content=data.get("content", ""),
             category=data.get("category", "memory"),
+            scope=data.get("scope", "global"),
+            scope_id=data.get("scope_id", ""),
             priority=data.get("priority", 0),
             created_at=data.get("created_at", time.time()),
             updated_at=data.get("updated_at", time.time()),
@@ -82,8 +88,8 @@ class MemoryStore:
             db = DBPools()
             async with db.sqlorContext("pipeline") as sor:
                 recs = await sor.sqlExe(
-                    "SELECT memory_key, content, category, priority, created_at, updated_at, access_count "
+                    "SELECT memory_key, content, category, scope, scope_id, priority, created_at, updated_at, access_count "
-                    "FROM pipeline_user_memory ORDER BY priority DESC, updated_at DESC LIMIT 200",
+                    "FROM pipeline_user_memory ORDER BY priority DESC, updated_at DESC LIMIT 500",
                     {},
                 )
                 for r in (recs or []):
@@ -91,6 +97,8 @@ class MemoryStore:
                         key=getattr(r, "memory_key", ""),
                         content=getattr(r, "content", ""),
                         category=getattr(r, "category", "memory"),
+                        scope=getattr(r, "scope", "global"),
+                        scope_id=getattr(r, "scope_id", ""),
                         priority=getattr(r, "priority", 0),
                         created_at=_ts(getattr(r, "created_at", None)),
                         updated_at=_ts(getattr(r, "updated_at", None)),
@@ -105,8 +113,12 @@ class MemoryStore:
         self._cache_loaded = True
 
     async def add(self, content: str, category: str = "memory",
-                  priority: int = PRIORITY_MEDIUM, key: str = None):
+                  priority: int = PRIORITY_MEDIUM, key: str = None,
-        """添加一条记忆。自动生成 key（取内容前80字符的哈希）。"""
+                  scope: str = "global", scope_id: str = ""):
+        """添加一条记忆。自动生成 key（取内容前80字符的哈希）。
+
+        scope 分域：global（通用）/ pipeline（产线）/ project（项目）/ user（用户个人）。
+        """
         await self._ensure_cache()
 
         if not key:
@@ -117,6 +129,8 @@ class MemoryStore:
             key=key,
             content=content,
             category=category,
+            scope=scope,
+            scope_id=scope_id,
             priority=priority,
             created_at=now,
             updated_at=now,
@@ -136,8 +150,12 @@ class MemoryStore:
         # 淘汰低优先级旧条目
         await self._evict(max_entries=100)
 
-    async def get(self, category: str = None, key: str = None) -> List[MemoryEntry]:
+    async def get(self, category: str = None, key: str = None,
-        """获取记忆。可按分类和 key 筛选。"""
+                  scope: str = None, scope_id: str = None) -> List[MemoryEntry]:
+        """获取记忆。可按分类/key/scope 筛选。
+
+        scope 过滤：scope=None 返回全部；scope 指定时返回该 scope 且 scope_id 匹配（或 scope_id=None 时该 scope 全部）。
+        """
         await self._ensure_cache()
 
         results = []
@@ -147,6 +165,11 @@ class MemoryStore:
                 for k, entry in self._cache[cat].items():
                     if key and k != key:
                         continue
+                    if scope is not None:
+                        if entry.scope != scope:
+                            continue
+                        if scope_id is not None and entry.scope_id != scope_id:
+                            continue
                     results.append(entry)
                     entry.access_count += 1
 
@@ -171,15 +194,30 @@ class MemoryStore:
         except Exception as e:
             logger.error(f"MemoryStore remove failed: {e}")
 
-    async def build_prompt_block(self, category: str = None, max_entries: int = 20) -> str:
+    async def build_prompt_block(self, category: str = None, max_entries: int = 20,
+                                 scope: str = None, scope_id: str = None) -> str:
         """构建注入 system prompt 的记忆段落。
 
+        指定 scope 时加载「global（通用）+ 指定 scope（专属）」的记忆，
+        实现通用记忆 + 产线/项目专属记忆的叠加注入。
         高优先级记忆注入完整内容，低优先级只注入摘要。
         """
-        entries = await self.get(category)
+        if scope:
+            # 通用 + 专属叠加
+            global_entries = await self.get(category, scope="global")
+            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)
+
         if not entries:
             return ""
 
+        entries.sort(key=lambda e: e.priority, reverse=True)
+
         blocks = []
         high_priority = [e for e in entries if e.priority >= self.PRIORITY_MEDIUM]
         low_priority = [e for e in entries if e.priority < self.PRIORITY_MEDIUM]
@@ -194,14 +232,15 @@ class MemoryStore:
         return "\n".join(blocks) if blocks else ""
 
     async def _db_upsert(self, entry: MemoryEntry):
-        """写入/更新 DB"""
+        """写入/更新 DB（同 key + category + scope + scope_id 才 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}$",
+                "SELECT id FROM pipeline_user_memory WHERE memory_key=${k}$ AND category=${c}$ "
-                {"k": entry.key, "c": entry.category},
+                "AND scope=${s}$ AND scope_id=${sid}$",
+                {"k": entry.key, "c": entry.category, "s": entry.scope, "sid": entry.scope_id},
             )
             if existing:
                 rid = existing[0].id
@@ -220,6 +259,8 @@ class MemoryStore:
                     "memory_key": entry.key,
                     "content": entry.content,
                     "category": entry.category,
+                    "scope": entry.scope,
+                    "scope_id": entry.scope_id,
                     "priority": entry.priority,
                 })