diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py
index 86f5bfb..ac9e085 100644
--- a/pipeline_service/agent_loop.py
+++ b/pipeline_service/agent_loop.py
@@ -54,6 +54,44 @@ def _normalize_role(role):
     return ROLE_ALIASES.get(r, r)
 
 
+async def _resolve_pipeline_id(project_id):
+    """从项目解析 pipeline_id（能力包 key），为空时 fallback 到默认产线。"""
+    pid = ""
+    try:
+        db = _get_db()
+        async with db.sqlorContext("pipeline") as sor:
+            recs = await sor.sqlExe(
+                "SELECT pipeline_id FROM sd_projects WHERE id=${pid}$", {"pid": project_id})
+            if recs:
+                pid = getattr(recs[0], "pipeline_id", "") or ""
+    except Exception:
+        pass
+    if not pid:
+        try:
+            from pipeline_core import DEFAULT_ABILITY_ID
+            pid = DEFAULT_ABILITY_ID
+        except ImportError:
+            pass
+    return pid
+
+
+async def _resolve_role(project_id, role):
+    """从能力包解析角色定义，返回 (normalized_role, role_specific_prompt, next_role)。
+
+    能力包未定义该角色时 fallback 到硬编码 ROLE_SPECIFICS/ROLE_CHAIN。
+    """
+    pid = await _resolve_pipeline_id(project_id)
+    try:
+        from pipeline_core import get_role_spec
+        spec = get_role_spec(pid, role)
+        if spec:
+            return spec.name, spec.system_prompt, spec.next_role
+    except Exception:
+        pass
+    norm = _normalize_role(role)
+    return norm, ROLE_SPECIFICS.get(norm, ROLE_SPECIFICS.get('develop', '')), ROLE_CHAIN.get(norm)
+
+
 def _get_db():
     from sqlor.dbpools import DBPools
     db = DBPools()
@@ -437,7 +475,17 @@ async def _get_repo_state(workspace_dir):
     return "\n".join(lines) if lines else "仓库无提交记录"
 
 
-async def _get_next_role(current_role):
+async def _get_next_role(current_role, project_id=""):
+    """任务链下一角色：优先从能力包取，fallback 硬编码 ROLE_CHAIN。"""
+    if project_id:
+        try:
+            pid = await _resolve_pipeline_id(project_id)
+            from pipeline_core import get_role_spec
+            spec = get_role_spec(pid, current_role)
+            if spec:
+                return spec.next_role  # 可能是 ""（终结）
+        except Exception:
+            pass
     return ROLE_CHAIN.get(_normalize_role(current_role))
 
 
@@ -599,7 +647,7 @@ async def _exec_agent_tool(tool, params, workspace_dir):
 # ── 角色 Agent ──
 
 async def role_agent_run(project_id, role, agent_id=None, model_name=None):
-    role = _normalize_role(role)
+    role, _role_specific, _ = await _resolve_role(project_id, role)
     if role == 'pm':
         return {"status": "idle", "message": "PM请使用 pm_review_run"}
 
@@ -629,7 +677,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
         except Exception as e:
             logger.warning(f"role_agent_run setup_repos failed: {e}")
 
-        role_specific = ROLE_SPECIFICS.get(role, ROLE_SPECIFICS.get('develop', ''))
+        role_specific = _role_specific
         qna_section = await _build_qna_section(sor, task_id)
         tools_text = json.dumps(AGENT_TOOLS, ensure_ascii=False)
 
@@ -893,7 +941,7 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
             })
             await sor.sqlExe("UPDATE pipeline_deliverables SET review_status='approved', review_comment=${cm}$ WHERE task_id=${tid}$", {"cm": comment, "tid": task_id})
             await sor.sqlExe("UPDATE pipeline_tasks SET state=${st}$, claimed_by=NULL WHERE id=${tid}$", {"st": TASK_APPROVED, "tid": task_id})
-            next_role = await _get_next_role(task_role)
+            next_role = await _get_next_role(task_role, project_id)
             if next_role:
                 next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment)
                 return {"status": "approved", "task_id": task_id, "next_task_id": next_tid, "next_role": next_role, "comment": comment}
diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py
index d21c302..532babf 100644
--- a/pipeline_service/agent_loop_v2.py
+++ b/pipeline_service/agent_loop_v2.py
@@ -355,9 +355,15 @@ class AgentExecutor:
             pass
         prompt += f"\n当前项目: {proj_name}\n" if proj_name else ""
 
-        # 注入记忆
+        # 注入记忆（分域：通用 + 产线 + 项目 叠加）
         if self.config.memory.enabled and self._memory_store:
-            mem_block = await self._memory_store.build_prompt_block(max_entries=15)
+            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
             if mem_block:
                 prompt += f"\n\n## 持久记忆\n{mem_block}"
 
diff --git a/pipeline_service/sdlc_ability.py b/pipeline_service/sdlc_ability.py
index ff837dd..239f8dc 100644
--- a/pipeline_service/sdlc_ability.py
+++ b/pipeline_service/sdlc_ability.py
@@ -18,6 +18,7 @@ import os
 from pipeline_core import (
     ToolDefinition,
     PipelineAbility,
+    RoleSpec,
     register_ability,
     SlashCommand,
     register_slash_command,
@@ -130,6 +131,86 @@ SDL_PROMPT = """你是「开发产线」的驾驶舱 agent，负责软件项目
 - 发现卡点（任务卡死、审核超时、失败、僵尸 claimed_by）时，主动定位根因并推动修复。"""
 
 
+# ── 产线角色集（可插拔，替代硬编码 ROLE_SPECIFICS/ROLE_ALIASES/ROLE_CHAIN） ──
+
+SDL_ROLES = [
+    RoleSpec(
+        name="requirement",
+        description="需求分析师",
+        aliases=["requirements", "requirement_analysis"],
+        system_prompt="""你是需求分析师。按 SDLC 仓库标准产出文档。
+
+产出路径: docs/00-requirement/requirement-spec.md
+同时创建（项目至少一个应用）:
+  - apps/<应用名>.md: 应用描述、部署环境、端口
+  - modules/<模块名>.md: 模块功能、仓库URL(先填写规划地址)、技术栈、依赖
+  - 每个应用至少关联一个模块
+内容: 项目概述、用户角色及权限、功能列表(每个功能:输入/处理/输出/验收标准)、非功能需求、业务流程。""",
+        next_role="design",
+    ),
+    RoleSpec(
+        name="design",
+        description="系统设计师",
+        aliases=["designer", "ui", "ux"],
+        system_prompt="""你是系统设计师。按 SDLC 仓库标准产出文档。
+
+产出路径: docs/01-design/ 目录下:
+  - architecture.md: 系统架构、技术选型理由
+  - database-design.md: ER图描述、表结构DDL
+  - api-design.md: 接口列表(method/path/request/response)
+  - ui-design.md: 页面结构、组件树(如适用)
+产出后用 result 输出文档，files 列出所有文件路径。""",
+        next_role="develop",
+    ),
+    RoleSpec(
+        name="develop",
+        description="开发工程师",
+        aliases=["developer", "dev", "development", "coding"],
+        system_prompt="""你是开发工程师。源码写入模块独立仓库，不在项目仓库。
+
+准备工作:
+  1. read_file 读 docs/01-design/ 下的设计文档
+  2. read_file 读 modules/<模块名>.md 获取模块仓库 URL
+  3. git_clone 克隆模块仓库到工作空间
+开发流程:
+  4. write_file 写代码到模块仓库目录下
+  5. run_shell 编译/运行验证
+  6. git_commit_push 提交 "develop: <简述>"
+  7. write_file 更新 modules/<模块名>.md 状态为已完成
+  8. write_file 写 docs/02-develop/dev-notes.md 记录开发内容
+  9. deliver 交付，files 列出所有产出文件路径
+PM审核: git_status 检查模块仓库有提交记录。""",
+        next_role="test",
+    ),
+    RoleSpec(
+        name="test",
+        description="测试工程师",
+        aliases=["testing", "qa", "tester"],
+        system_prompt="""你是测试工程师。按 SDLC 仓库标准产出文档。
+
+产出路径: docs/03-test/ 目录下:
+  - test-plan.md: 测试策略(单元/集成/端到端)
+  - test-cases.md: 用例清单(编号/前置条件/步骤/预期结果)
+  - test-report.md: 执行结果、Bug清单、覆盖率
+测试脚本放到 files。""",
+        next_role="deploy",
+    ),
+    RoleSpec(
+        name="deploy",
+        description="部署运维工程师",
+        aliases=["deployment", "release"],
+        system_prompt="""你是部署运维工程师。按 SDLC 仓库标准产出文档和配置。
+
+产出路径:
+  - docs/04-deploy/deploy-guide.md: 环境要求、部署步骤、回滚方案
+  - docs/04-deploy/release-notes.md: 发布说明
+  - config/ 下: Dockerfile、nginx配置、docker-compose.yml 等
+产出后用 result 输出部署说明，files 列出所有配置文件路径。""",
+        next_role="",
+    ),
+]
+
+
 # ── handler 独立函数（签名统一 async def handler(sor, params, ctx) -> str） ──
 
 async def _h_create_task(sor, p, ctx):
@@ -528,6 +609,7 @@ def register_sdlc_ability():
         tools=SDL_TOOLS,
         system_prompt=SDL_PROMPT,
         handlers=SDL_HANDLERS,
+        roles=SDL_ROLES,
     )
     register_ability(ability)
     return ability