diff --git a/pipeline_core/__init__.py b/pipeline_core/__init__.py
index 84033e8..60ad6fd 100644
--- a/pipeline_core/__init__.py
+++ b/pipeline_core/__init__.py
@@ -36,6 +36,14 @@ from .ability import (
     get_ability_tools,
     get_ability_handler,
 )
+from .slash import (
+    SlashCommand,
+    register_slash_command,
+    get_slash_commands,
+    resolve_slash_command,
+    parse_slash_args,
+    list_commands,
+)
 from .tool_registry import (
     ToolRegistry,
     get_tool_registry,
diff --git a/pipeline_core/agent_config.py b/pipeline_core/agent_config.py
index 7b3727c..de23bd7 100644
--- a/pipeline_core/agent_config.py
+++ b/pipeline_core/agent_config.py
@@ -49,12 +49,15 @@ class MemoryConfig:
 
 @dataclass
 class SkillConfig:
-    """技能配置 — 三级隔离：全局 + 组织 + 用户"""
+    """技能配置 — 六级隔离：global→org→pipeline→role→project→user"""
     enabled: bool = True
-    base_dir: str = "skills"       # 技能根目录，其下为 global/orgs/users/
+    base_dir: str = "skills"       # 技能根目录，其下为 global/pipelines/projects/orgs/users/
     max_skills_per_turn: int = 5
     enable_global: bool = True     # 启用全局技能
     enable_org: bool = True        # 启用组织技能
+    enable_pipeline: bool = True   # 启用产线技能
+    enable_role: bool = True       # 启用角色技能
+    enable_project: bool = True    # 启用项目/会话技能
     enable_user: bool = True       # 启用用户技能
 
 
@@ -117,6 +120,9 @@ class AgentConfig:
                 "base_dir": self.skills.base_dir,
                 "enable_global": self.skills.enable_global,
                 "enable_org": self.skills.enable_org,
+                "enable_pipeline": self.skills.enable_pipeline,
+                "enable_role": self.skills.enable_role,
+                "enable_project": self.skills.enable_project,
                 "enable_user": self.skills.enable_user,
             },
             "tools": [
@@ -165,6 +171,9 @@ class AgentConfig:
                 max_skills_per_turn=sk.get("max_skills_per_turn", 5),
                 enable_global=sk.get("enable_global", True),
                 enable_org=sk.get("enable_org", True),
+                enable_pipeline=sk.get("enable_pipeline", True),
+                enable_role=sk.get("enable_role", True),
+                enable_project=sk.get("enable_project", True),
                 enable_user=sk.get("enable_user", True),
             ),
             tools=[ToolDefinition(**t) if isinstance(t, dict) else t for t in data.get("tools", [])],
diff --git a/pipeline_core/skill_loader.py b/pipeline_core/skill_loader.py
index f33a8d2..fbf7a60 100644
--- a/pipeline_core/skill_loader.py
+++ b/pipeline_core/skill_loader.py
@@ -1,33 +1,55 @@
 """
-pipeline-core: Skill Loader v2 — 三级技能隔离
+pipeline-core: Skill Loader — 六级可插拔技能隔离
+
+技能目录结构（每级可独立存在，同名高优先级覆盖低优先级）：
 
-技能目录结构：
   skills/
-    global/              ← 全局技能（所有人可见）
-      skill-name/
-        SKILL.md
-    orgs/                ← 组织技能
-      {org_id}/
-        skill-name/
-          SKILL.md
-    users/               ← 用户个人技能
-      {user_id}/
-        skill-name/
-          SKILL.md
+    global/                          ← 通用技能（core 内核，所有产线所有角色）
+      skill-name/SKILL.md
+    pipelines/                       ← 产线技能
+      {pipeline_id}/
+        common/                      ← 产线通用技能（该产线所有角色可见）
+          skill-name/SKILL.md
+        roles/                       ← 角色技能（产线内角色专属）
+          {role}/
+            skill-name/SKILL.md
+    projects/                        ← 项目/会话技能（临时、项目特定）
+      {project_id}/
+        skill-name/SKILL.md
+    orgs/{org_id}/                   ← 组织技能
+    users/{user_id}/                 ← 用户个人技能
 
-加载优先级：用户 > 组织 > 全局
-同名技能：用户级覆盖组织级覆盖全局级
+加载优先级（低 → 高，同名后者覆盖前者）：
+  global → org → pipeline(common) → role → project → user
 """
 
-import json
 import logging
 import os
-import re
 from dataclasses import dataclass, field
 from typing import Dict, List, Optional
 
 logger = logging.getLogger("pipeline.skill_loader")
 
+# scope 优先级（数值越大优先级越高）
+SCOPE_PRIORITY = {
+    "global": 0,
+    "org": 1,
+    "pipeline": 2,
+    "role": 3,
+    "project": 4,
+    "user": 5,
+}
+
+SCOPE_TAG = {
+    "global": "通用",
+    "org": "组织",
+    "pipeline": "产线",
+    "role": "角色",
+    "project": "项目",
+    "user": "个人",
+}
+
+
 # ── 技能数据模型 ──
 
 @dataclass
@@ -37,15 +59,14 @@ class Skill:
     path: str                        # SKILL.md 绝对路径
     description: str = ""
     content: str = ""
-    scope: str = "global"            # global / org / user
-    scope_id: str = ""               # org_id 或 user_id
+    scope: str = "global"            # global/org/pipeline/role/project/user
+    scope_id: str = ""               # org_id / pipeline_id / pipeline_id:role / project_id / user_id
     trigger_keywords: List[str] = field(default_factory=list)
     version: str = "1.0.0"
 
     @classmethod
     def from_file(cls, filepath: str, scope: str = "global",
                   scope_id: str = "") -> "Skill":
-        """从 SKILL.md 文件加载技能"""
         name = os.path.basename(os.path.dirname(filepath))
         content = ""
         description = ""
@@ -59,13 +80,11 @@ class Skill:
             logger.warning(f"Skill read failed: {filepath} err={e}")
             return cls(name=name, path=filepath, scope=scope, scope_id=scope_id)
 
-        # 解析 YAML frontmatter
         fm = _parse_frontmatter(content)
         if fm:
             description = fm.get("description", description)
             trigger_keywords = fm.get("trigger_keywords", trigger_keywords)
             version = fm.get("version", version)
-            # frontmatter 中的 scope 优先
             scope = fm.get("scope", scope)
 
         if not description:
@@ -82,8 +101,7 @@ class Skill:
         )
 
     def to_prompt_block(self) -> str:
-        """生成注入到 system prompt 的技能文本块"""
-        scope_tag = {"global": "全局", "org": "组织", "user": "个人"}.get(self.scope, "")
+        scope_tag = SCOPE_TAG.get(self.scope, self.scope)
         return f"""## [{scope_tag}] {self.name}
 {self.description}
 
@@ -92,7 +110,6 @@ class Skill:
 
 
 def _parse_frontmatter(content: str) -> Optional[dict]:
-    """解析 YAML frontmatter"""
     if not content.startswith("---"):
         return None
     end = content.find("---", 3)
@@ -113,22 +130,23 @@ def _parse_frontmatter(content: str) -> Optional[dict]:
     return result
 
 
-# ── 三级技能加载器 ──
+# ── 六级技能加载器 ──
 
 class SkillLoader:
-    """三级技能加载器：global → org → user"""
+    """六级技能加载器：global → org → pipeline → role → project → user"""
 
     def __init__(self, base_dir: str = ""):
         self.base_dir = base_dir
-        # 三级存储: {scope: {name: Skill}}
         self._global: Dict[str, Skill] = {}
-        self._orgs: Dict[str, Dict[str, Skill]] = {}     # {org_id: {name: Skill}}
-        self._users: Dict[str, Dict[str, Skill]] = {}    # {user_id: {name: Skill}}
+        self._orgs: Dict[str, Dict[str, Skill]] = {}        # {org_id: {name: Skill}}
+        self._pipelines: Dict[str, Dict[str, Skill]] = {}   # {pipeline_id: {name: Skill}}
+        self._roles: Dict[str, Dict[str, Skill]] = {}       # {pipeline_id:role: {name: Skill}}
+        self._projects: Dict[str, Dict[str, Skill]] = {}    # {project_id: {name: Skill}}
+        self._users: Dict[str, Dict[str, Skill]] = {}       # {user_id: {name: Skill}}
         if base_dir:
             self.reload()
 
     def set_base_dir(self, base_dir: str):
-        """设置技能根目录并扫描"""
         self.base_dir = base_dir
         if base_dir:
             self.reload()
@@ -136,41 +154,74 @@ class SkillLoader:
     # ── 目录扫描 ──
 
     def reload(self):
-        """重新扫描所有技能目录"""
         self._global = {}
         self._orgs = {}
+        self._pipelines = {}
+        self._roles = {}
+        self._projects = {}
         self._users = {}
 
         if not self.base_dir or not os.path.isdir(self.base_dir):
             return
 
-        # 全局技能: skills/global/
+        # 通用: skills/global/
         global_dir = os.path.join(self.base_dir, "global")
         if os.path.isdir(global_dir):
             self._global = self._scan_scope_dir(global_dir, "global")
 
-        # 组织技能: skills/orgs/{org_id}/
+        # 组织: skills/orgs/{org_id}/
         orgs_dir = os.path.join(self.base_dir, "orgs")
         if os.path.isdir(orgs_dir):
             for org_id in os.listdir(orgs_dir):
-                org_path = os.path.join(orgs_dir, org_id)
-                if os.path.isdir(org_path):
-                    self._orgs[org_id] = self._scan_scope_dir(org_path, "org", org_id)
+                p = os.path.join(orgs_dir, org_id)
+                if os.path.isdir(p):
+                    self._orgs[org_id] = self._scan_scope_dir(p, "org", org_id)
 
-        # 用户技能: skills/users/{user_id}/
+        # 产线: skills/pipelines/{pipeline_id}/common/ + roles/{role}/
+        pipelines_dir = os.path.join(self.base_dir, "pipelines")
+        if os.path.isdir(pipelines_dir):
+            for pid in os.listdir(pipelines_dir):
+                pdir = os.path.join(pipelines_dir, pid)
+                if not os.path.isdir(pdir):
+                    continue
+                # 产线通用技能
+                common_dir = os.path.join(pdir, "common")
+                if os.path.isdir(common_dir):
+                    self._pipelines[pid] = self._scan_scope_dir(common_dir, "pipeline", pid)
+                # 角色技能
+                roles_dir = os.path.join(pdir, "roles")
+                if os.path.isdir(roles_dir):
+                    for role in os.listdir(roles_dir):
+                        rdir = os.path.join(roles_dir, role)
+                        if os.path.isdir(rdir):
+                            key = f"{pid}:{role}"
+                            self._roles[key] = self._scan_scope_dir(rdir, "role", key)
+
+        # 项目/会话: skills/projects/{project_id}/
+        projects_dir = os.path.join(self.base_dir, "projects")
+        if os.path.isdir(projects_dir):
+            for project_id in os.listdir(projects_dir):
+                p = os.path.join(projects_dir, project_id)
+                if os.path.isdir(p):
+                    self._projects[project_id] = self._scan_scope_dir(p, "project", project_id)
+
+        # 用户: skills/users/{user_id}/
         users_dir = os.path.join(self.base_dir, "users")
         if os.path.isdir(users_dir):
             for user_id in os.listdir(users_dir):
-                user_path = os.path.join(users_dir, user_id)
-                if os.path.isdir(user_path):
-                    self._users[user_id] = self._scan_scope_dir(user_path, "user", user_id)
+                p = os.path.join(users_dir, user_id)
+                if os.path.isdir(p):
+                    self._users[user_id] = self._scan_scope_dir(p, "user", user_id)
 
         logger.debug(f"SkillLoader reloaded: global={len(self._global)}, "
-                     f"orgs={len(self._orgs)}, users={len(self._users)}")
+                     f"orgs={sum(len(s) for s in self._orgs.values())}, "
+                     f"pipelines={sum(len(s) for s in self._pipelines.values())}, "
+                     f"roles={sum(len(s) for s in self._roles.values())}, "
+                     f"projects={sum(len(s) for s in self._projects.values())}, "
+                     f"users={sum(len(s) for s in self._users.values())}")
 
     def _scan_scope_dir(self, scope_dir: str, scope: str,
                         scope_id: str = "") -> Dict[str, Skill]:
-        """扫描单个 scope 目录下的所有技能"""
         skills = {}
         for entry in os.listdir(scope_dir):
             entry_path = os.path.join(scope_dir, entry)
@@ -180,7 +231,6 @@ class SkillLoader:
             if os.path.isfile(skill_file):
                 skill = Skill.from_file(skill_file, scope, scope_id)
                 skills[skill.name] = skill
-            # 也支持子目录嵌套
             for sub in os.listdir(entry_path):
                 sub_path = os.path.join(entry_path, sub)
                 if os.path.isdir(sub_path):
@@ -192,24 +242,40 @@ class SkillLoader:
 
     # ── 查询 ──
 
-    def get_merged(self, org_id: str = "", user_id: str = "") -> Dict[str, Skill]:
-        """按优先级合并：全局 + 组织 + 用户。同名技能后者覆盖前者。"""
-        merged = dict(self._global)  # 全局为基础
+    def get_merged(self, pipeline_id: str = "", role: str = "",
+                   project_id: str = "", org_id: str = "",
+                   user_id: str = "") -> Dict[str, Skill]:
+        """按优先级合并六级技能。同名技能高优先级覆盖低优先级。
+
+        优先级（低→高）：global → org → pipeline → role → project → user
+        """
+        merged = dict(self._global)
 
-        # 组织技能覆盖
         if org_id and org_id in self._orgs:
             merged.update(self._orgs[org_id])
 
-        # 用户技能覆盖（最高优先级）
+        if pipeline_id and pipeline_id in self._pipelines:
+            merged.update(self._pipelines[pipeline_id])
+
+        if pipeline_id and role:
+            role_key = f"{pipeline_id}:{role}"
+            if role_key in self._roles:
+                merged.update(self._roles[role_key])
+
+        if project_id and project_id in self._projects:
+            merged.update(self._projects[project_id])
+
         if user_id and user_id in self._users:
             merged.update(self._users[user_id])
 
         return merged
 
-    def get_by_trigger(self, user_input: str, org_id: str = "",
-                       user_id: str = "", max_skills: int = 5) -> List[Skill]:
-        """根据用户输入匹配相关技能（基于 trigger_keywords）"""
-        all_skills = list(self.get_merged(org_id, user_id).values())
+    def get_by_trigger(self, user_input: str, pipeline_id: str = "",
+                       role: str = "", project_id: str = "",
+                       org_id: str = "", user_id: str = "",
+                       max_skills: int = 5) -> List[Skill]:
+        """根据用户输入匹配相关技能（trigger_keywords + scope 加权）"""
+        all_skills = list(self.get_merged(pipeline_id, role, project_id, org_id, user_id).values())
         scored = []
         user_lower = user_input.lower()
         for skill in all_skills:
@@ -219,45 +285,41 @@ class SkillLoader:
                     score += 1
             if skill.name.lower() in user_lower:
                 score += 2
-            # scope 加分：用户 > 组织 > 全局
-            if skill.scope == "user":
-                score += 3
-            elif skill.scope == "org":
-                score += 2
-            elif skill.scope == "global":
-                score += 1
+            # scope 优先级加权
+            score += SCOPE_PRIORITY.get(skill.scope, 0)
             if score > 0:
                 scored.append((score, skill))
         scored.sort(key=lambda x: x[0], reverse=True)
         return [s[1] for s in scored[:max_skills]]
 
-    def get_skill_count(self, org_id: str = "", user_id: str = "") -> int:
-        """可见技能总数"""
-        return len(self.get_merged(org_id, user_id))
+    def get_skill_count(self, pipeline_id: str = "", role: str = "",
+                        project_id: str = "", org_id: str = "",
+                        user_id: str = "") -> int:
+        return len(self.get_merged(pipeline_id, role, project_id, org_id, user_id))
 
     def list_by_scope(self) -> Dict[str, int]:
-        """各 scope 的技能数量"""
         return {
             "global": len(self._global),
             "orgs": sum(len(s) for s in self._orgs.values()),
+            "pipelines": sum(len(s) for s in self._pipelines.values()),
+            "roles": sum(len(s) for s in self._roles.values()),
+            "projects": sum(len(s) for s in self._projects.values()),
             "users": sum(len(s) for s in self._users.values()),
         }
 
     # ── Prompt 构建 ──
 
-    def build_prompt_block(self, org_id: str = "", user_id: str = "",
-                           user_input: str = None, max_skills: int = 5) -> str:
-        """构建注入 system prompt 的技能段落。
-
-        有 user_input 时按 trigger 匹配，否则取全部（受限 max_skills）。
-        """
+    def build_prompt_block(self, pipeline_id: str = "", role: str = "",
+                           project_id: str = "", org_id: str = "",
+                           user_id: str = "", user_input: Optional[str] = None,
+                           max_skills: int = 5) -> str:
         if user_input:
-            skills = self.get_by_trigger(user_input, org_id, user_id, max_skills)
+            skills = self.get_by_trigger(user_input, pipeline_id, role,
+                                         project_id, org_id, user_id, max_skills)
         else:
-            merged = self.get_merged(org_id, user_id)
-            # 优先用户级
+            merged = self.get_merged(pipeline_id, role, project_id, org_id, user_id)
             skills = sorted(merged.values(),
-                            key=lambda s: {"user": 0, "org": 1, "global": 2}.get(s.scope, 9))
+                            key=lambda s: SCOPE_PRIORITY.get(s.scope, 9))
             skills = skills[:max_skills]
 
         if not skills:
@@ -265,14 +327,15 @@ class SkillLoader:
 
         blocks = ["## 可用技能\n"]
         for s in skills:
-            blocks.append(f"- **[{s.scope}] {s.name}**: {s.description[:200]}")
+            blocks.append(f"- **[{SCOPE_TAG.get(s.scope, s.scope)}] {s.name}**: {s.description[:200]}")
         blocks.append("")
         return "\n".join(blocks)
 
     def build_full_prompt_block(self, skill_names: List[str],
-                                org_id: str = "", user_id: str = "") -> str:
-        """注入指定技能的完整内容"""
-        merged = self.get_merged(org_id, user_id)
+                                pipeline_id: str = "", role: str = "",
+                                project_id: str = "", org_id: str = "",
+                                user_id: str = "") -> str:
+        merged = self.get_merged(pipeline_id, role, project_id, org_id, user_id)
         blocks = []
         for name in skill_names:
             if name in merged:
@@ -286,7 +349,6 @@ _loader: Optional[SkillLoader] = None
 
 
 def get_skill_loader(base_dir: str = "") -> SkillLoader:
-    """获取全局技能加载器单例。首次调用可传入 base_dir。"""
     global _loader
     if _loader is None:
         _loader = SkillLoader(base_dir)
diff --git a/pipeline_core/slash.py b/pipeline_core/slash.py
new file mode 100644
index 0000000..a520d37
--- /dev/null
+++ b/pipeline_core/slash.py
@@ -0,0 +1,83 @@
+"""
+pipeline-core: SlashCommand — 可插拔 slash 命令注册表
+
+slash 命令 = 产线/角色/通用 的快捷指令，用户以 "/" 开头输入时直接触发，
+不进入 LLM 循环（类似 Hermes CLI 的 /new /model /tools /skills）。
+
+命令可插拔：通用命令在 core 注册，产线命令在能力包（sdlc_ability）注册，
+不同产线可有不同的 slash 命名。
+
+依赖方向：pipeline-core（本文件，纯定义 + 注册表）← pipeline-service（注册通用命令）
+          ← sdlc_ability（注册 SDLC 命令）
+"""
+
+import logging
+from dataclasses import dataclass, field
+from typing import Callable, Dict, List, Optional
+
+logger = logging.getLogger("pipeline.slash")
+
+
+@dataclass
+class SlashCommand:
+    """单个 slash 命令。handler 签名：async def handler(args, ctx) -> str
+      - args: 命令后的参数字符串（不含命令名）
+      - ctx: AgentExecutor 上下文 dict（project_id/user_id/pipeline_id/workspace_dir/config/executor）
+    """
+    name: str                              # 不含 "/" 前缀，如 "new"
+    description: str = ""
+    handler: Optional[Callable] = None
+    scope: str = "global"                  # global / pipeline / role
+    scope_id: str = ""                     # pipeline_id 或 pipeline_id:role
+    args_hint: str = ""                    # 参数提示，如 "[task_id]"
+
+
+# ── 全局注册表 ──
+
+_commands: Dict[str, SlashCommand] = {}
+
+
+def register_slash_command(cmd: SlashCommand):
+    """注册一个 slash 命令（幂等：重复注册覆盖）。"""
+    _commands[cmd.name] = cmd
+    logger.debug(f"SlashCommand registered: /{cmd.name} ({cmd.scope})")
+
+
+def get_slash_commands(pipeline_id: str = "", role: str = "") -> Dict[str, SlashCommand]:
+    """取当前上下文可见的 slash 命令（通用 + 产线 + 角色）。"""
+    result: Dict[str, SlashCommand] = {}
+    for name, cmd in _commands.items():
+        if cmd.scope == "global":
+            result[name] = cmd
+        elif cmd.scope == "pipeline" and pipeline_id and cmd.scope_id == pipeline_id:
+            result[name] = cmd
+        elif cmd.scope == "role" and pipeline_id and role \
+                and cmd.scope_id == f"{pipeline_id}:{role}":
+            result[name] = cmd
+    return result
+
+
+def resolve_slash_command(text: str, pipeline_id: str = "",
+                          role: str = "") -> Optional[SlashCommand]:
+    """从用户输入解析 slash 命令（text 以 "/" 开头）。"""
+    if not text or not text.startswith("/"):
+        return None
+    body = text[1:].strip()
+    if not body:
+        return None
+    # 命令名 = 第一个空格前的 token
+    name = body.split()[0] if " " in body else body
+    visible = get_slash_commands(pipeline_id, role)
+    return visible.get(name)
+
+
+def parse_slash_args(text: str) -> str:
+    """提取命令名之后的参数字符串。"""
+    body = text[1:].strip() if text.startswith("/") else text.strip()
+    parts = body.split(None, 1)
+    return parts[1] if len(parts) > 1 else ""
+
+
+def list_commands() -> List[SlashCommand]:
+    """列出所有已注册命令。"""
+    return list(_commands.values())