feat: role-based skill loading — guess role from task title, load common/{role} skills

wwwroot/api/cockpit_chat.dspy

@@ -25,27 +25,57 @@ DEFAULT_SYSTEM_PROMPT = """你是一个专业的软件开发 Agent，名为「
 - 环境部署和验证"""
 
 
-def _load_skills(skills_dir):
+def _guess_role(title):
-    """Scan skills_dir for SKILL.md files and return their content."""
+    """Guess agent role from task title keywords."""
+    title_lower = (title or '').lower()
+    keywords = {
+        'design': ['设计', 'design', '架构', '方案', '需求', '原型', 'ui', 'ux'],
+        'develop': ['开发', '编码', '实现', '编写', 'develop', 'code', 'build', '重构', '修复'],
+        'test': ['测试', 'test', '验证', '检查', 'review', '评审'],
+        'deploy': ['部署', '发布', 'deploy', 'release', '上线', '配置'],
+        'ops': ['运维', '监控', 'ops', '日志', '备份', '迁移'],
+    }
+    for role, kws in keywords.items():
+        for kw in kws:
+            if kw in title_lower:
+                return role
+    return 'develop'  # default
+
+
+def _load_skills(skills_dir, role=None):
+    """Scan skills_dir for SKILL.md files. If role given, load common/ + {role}/ only."""
     skills = []
     if not skills_dir:
         return skills
-    try:
+    dirs_to_scan = ['common']
-        for name in os.listdir(skills_dir):
+    if role:
-            skill_path = os.path.join(skills_dir, name)
+        dirs_to_scan.append(role)
-            skill_md = os.path.join(skill_path, 'SKILL.md')
+    else:
-            if os.path.isdir(skill_path) and os.path.isfile(skill_md):
+        # No role specified — scan all subdirs
-                try:
+        try:
-                    with open(skill_md, 'r') as f:
+            dirs_to_scan = [d for d in os.listdir(skills_dir)
-                        content = f.read()
+                           if os.path.isdir(os.path.join(skills_dir, d))]
-                    # Truncate very large skills
+        except Exception:
-                    if len(content) > 8000:
+            return skills
-                        content = content[:8000] + '\n\n... (truncated)'
+    for subdir in dirs_to_scan:
-                    skills.append({'name': name, 'content': content})
+        subdir_path = os.path.join(skills_dir, subdir)
-                except Exception:
+        if not os.path.isdir(subdir_path):
-                    pass
+            continue
-    except Exception:
+        try:
-        pass
+            for name in os.listdir(subdir_path):
+                skill_path = os.path.join(subdir_path, name)
+                skill_md = os.path.join(skill_path, 'SKILL.md')
+                if os.path.isdir(skill_path) and os.path.isfile(skill_md):
+                    try:
+                        with open(skill_md, 'r') as f:
+                            content = f.read()
+                        if len(content) > 8000:
+                            content = content[:8000] + '\n\n... (truncated)'
+                        skills.append({'name': name, 'role': subdir, 'content': content})
+                    except Exception:
+                        pass
+        except Exception:
+            pass
     return skills
 
 
@@ -482,9 +512,11 @@ if action == 'send_message':
 
                     for t in tasks:
                         try:
-                            # Build system prompt with skills injected
+                            # Guess role and load role-specific skills
+                            role = _guess_role(getattr(t, 'title', ''))
+                            task_skills = _load_skills(ctx.get('skills_dir', ''), role)
                             sp = settings['system_prompt']
-                            skills_prompt = _build_skills_prompt(ctx.get('skills', []))
+                            skills_prompt = _build_skills_prompt(task_skills)
                             if skills_prompt:
                                 sp = sp + skills_prompt
                             task_msgs = [{"role": "system", "content": sp}]