feat: 任务来源分类机制 + classify_task 工具 + 模块技能注入项目空间

根因修复（develop 不调 fix_bug 导致 bug 恒 open、链断）：
1. 三条任务创建路径打 task_kind 标记：_create_next_task=new_dev、
   _rollback_task_chain=rework、_pm_create_tasks 按标题判 bug_fix/new_dev
2. bug_capability 新增 classify_task 工具：读 task_kind 标记判断任务来源
   （老任务按 rollback_from/title 兜底），develop 据此决定是否走 fix_bug 状态机
3. TOOL_SCHEMAS 注册 classify_task + capability_ctx 注入 task_id
4. 增强能力：_collect_module_skills 运行时扫描 workspace repos/*/skill/SKILL.md，
   作为「项目模块」scope 注入技能目录 + load_skill 支持加载模块技能全文，
   让项目角色知道引用的业务模块怎么用（架构/数据模型/挂载函数/坑）

pipeline_service/agent_loop.py

@@ -623,6 +623,66 @@ async def _get_workspace_dir(sor, project_id):
     return _resolve_workspace(ws)
 
 
+def _parse_skill_frontmatter(content):
+    """解析技能 SKILL.md 的 frontmatter（--- 之间），返回 dict。无 frontmatter 返回 {}。"""
+    content = (content or "").strip()
+    if not content.startswith("---"):
+        return {}
+    end = content.find("---", 3)
+    if end == -1:
+        return {}
+    fm = {}
+    for line in content[3:end].strip().split("\n"):
+        line = line.strip()
+        if ":" in line:
+            k, _, v = line.partition(":")
+            k = k.strip()
+            v = v.strip().strip('"').strip("'")
+            if v.startswith("[") and v.endswith("]"):
+                v = [x.strip().strip('"').strip("'") for x in v[1:-1].split(",") if x.strip()]
+            fm[k] = v
+    return fm
+
+
+def _collect_module_skills(workspace_dir):
+    """扫描项目 workspace 的 repos/*/skill/SKILL.md，收集模块技能（模块怎么用）。
+
+    模块仓库自带 skill/SKILL.md（架构/数据模型/挂载函数/坑位），但这些不进 skill_loader
+    的静态文件树，导致项目角色不知道模块怎么用。这里运行时扫描，作为「项目模块」scope
+    注入技能目录 + 支持 load_skill 加载全文。返回 [{name, description, content, path}]。
+    """
+    modules = []
+    repos_dir = os.path.join(workspace_dir, 'repos')
+    if not os.path.isdir(repos_dir):
+        return modules
+    for entry in sorted(os.listdir(repos_dir)):
+        skill_file = os.path.join(repos_dir, entry, 'skill', 'SKILL.md')
+        if not os.path.isfile(skill_file):
+            continue
+        try:
+            with open(skill_file, 'r', encoding='utf-8') as f:
+                content = f.read()
+        except Exception:
+            continue
+        fm = _parse_skill_frontmatter(content)
+        name = (fm.get('name') or entry).strip()
+        description = (fm.get('description') or '').strip()
+        if not description:
+            for line in content.split("\n"):
+                line = line.strip()
+                if line and not line.startswith("#") and not line.startswith("---"):
+                    description = line[:200]
+                    break
+        modules.append({
+            "name": name,
+            "description": description,
+            "content": content,
+            "path": skill_file,
+            "repo": entry,
+        })
+    return modules
+
+
 async def _build_qna_section(sor, task_id, role, agent_id=None):
     from .communication import get_task_qa
     qa = await get_task_qa(task_id, role=role, agentid=agent_id)
@@ -780,6 +840,9 @@ async def _create_next_task(sor, project_id, task, next_role, pm_comment=''):
     new_title = f"{title}（{next_role}阶段）"
     new_params = {**params, 'previous_role': _normalize_role(getattr(task, 'role', '')),
                   'previous_task_id': getattr(task, 'id', ''), 'pm_comment': pm_comment}
+    # 任务来源标记：系统自动派生（design→develop→deploy_test→test）都是「新开发」链，
+    # 不含 bug 修复语义。develop 角色据此判断是否要走 fix_bug 状态机。
+    new_params['task_kind'] = 'new_dev'
     # 清除 pm_assigned：它是「PM 按模块清单派发」的标记，只作用于当前任务；
     # 下一角色任务是系统自动创建（非 PM 派发），若继承会污染——例如 PM 派发的 design 任务
     # 带 pm_assigned=True，design approved 后自动创建的 develop 任务继承了它，被 1785 行的
@@ -859,6 +922,8 @@ async def _rollback_task_chain(sor, project_id, task_id, rollback_role, comment)
     except (json.JSONDecodeError, TypeError):
         tp = {}
     new_params = {**tp, 'rollback_from': task_id, 'rollback_comment': comment}
+    # 任务来源标记：回退重做 = 修复导致上游失败的缺陷（质检重做），develop 据此必须走 bug 状态机。
+    new_params['task_kind'] = 'rework'
     # 清 pm_assigned：回退重做任务是系统重建的任务（非 PM 按模块清单派发），继承 target_task 的
     # pm_assigned 会污染——若 target 是应用级 develop（历史脏数据带 pm_assigned=True），approve 后被
     # 「模块级 develop → 跳过 deploy_test」误判，任务链断在 approved。与 _create_next_task 的 pop 对齐。
@@ -1088,8 +1153,6 @@ async def _build_role_skills_block(sor, project_id, role, org_id=""):
         loader = get_skill_loader(skills_dir)
         merged = loader.get_merged(pipeline_id=pid, role=role,
                                    project_id=project_id, org_id=org_id or '0')
-        if not merged:
-            return ""
         skills = sorted(merged.values(),
                         key=lambda s: (-int(getattr(s, 'essential', False)),
                                        -SCOPE_PRIORITY.get(getattr(s, 'scope', ''), 0)))
@@ -1097,6 +1160,17 @@ async def _build_role_skills_block(sor, project_id, role, org_id=""):
         for s in skills[:60]:
             tag = SCOPE_TAG.get(getattr(s, 'scope', ''), getattr(s, 'scope', ''))
             lines.append(f"- [{tag}] {s.name}: {(s.description or '')[:100]}")
+        # 项目模块技能：workspace repos/*/skill/SKILL.md（模块怎么用——架构/数据模型/挂载函数/坑位），
+        # 运行时扫描注入，让项目角色知道引用了哪些模块、每个模块怎么挂载（load_xxx 入口/库名/坑）。
+        try:
+            ws = await _get_workspace_dir(sor, project_id)
+            module_skills = _collect_module_skills(ws)
+            if module_skills:
+                lines.append("\n## 项目模块技能（本项目引用的业务模块，load_skill 加载全文了解模块怎么用）")
+                for m in module_skills:
+                    lines.append(f"- [项目模块] {m['name']}: {(m['description'] or '')[:100]}")
+        except Exception as e:
+            logger.warning(f"module skills collect failed: {e}")
         return "\n".join(lines)
     except Exception as e:
         logger.warning(f"role skills load failed: {e}")
@@ -1154,6 +1228,14 @@ async def _load_skill_by_name(sor, project_id, role, org_id, name, file_path=Non
         merged = loader.get_merged(pipeline_id=pid, role=role,
                                    project_id=project_id, org_id=org_id or '0')
         if name not in merged:
+            # 项目模块技能（workspace repos/*/skill/SKILL.md）不在 skill_loader 静态树里，运行时补查
+            try:
+                ws = await _get_workspace_dir(sor, project_id)
+                for m in _collect_module_skills(ws):
+                    if m['name'] == name or m['repo'] == name:
+                        return f"## [项目模块] {m['name']}\n{m['description']}\n\n{m['content']}"
+            except Exception:
+                pass
             names = ", ".join(sorted(merged.keys())) or "（无可用技能）"
             return f"FAIL: 技能 '{name}' 不存在。可用技能: {names}"
         skill = merged[name]
@@ -1237,6 +1319,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
             "iteration_id": _iter.get('id', '') if _iter else '',
             "who": role,
             "agent_id": agent_id,
+            "task_id": task_id,
         }
 
         system = (AGENT_SYSTEM_PROMPT
@@ -1483,6 +1566,13 @@ async def _pm_create_tasks(sor, project_id, params, parent_task_id=None):
             cancelled.append(did)
 
         tparams = {"description": desc, "pm_assigned": True}
+        # 任务来源标记：PM 派发的任务按 title/description 判断——
+        # 含「修复 Bug」= bug_fix（develop 必须走 fix_bug 状态机）；否则 = new_dev（模块开发等）。
+        _title_desc = f"{title} {desc}"
+        if '修复 Bug' in _title_desc or '修复Bug' in _title_desc or '修复bug' in _title_desc:
+            tparams['task_kind'] = 'bug_fix'
+        else:
+            tparams['task_kind'] = 'new_dev'
         if iteration_name:
             tparams["iteration_id"] = iteration_name
         tid = getID()

pipeline_service/bug_capability.py

@@ -11,6 +11,7 @@ scope 约定：sd_bugs 无 project_id 列，用 iteration_id 做范围校验（C
 """
 
 import logging
+import json
 from datetime import datetime
 from sqlor.dbpools import DBPools
 from appPublic.uniqueID import getID
@@ -223,3 +224,46 @@ def _rec_to_dict(rec):
             except Exception:
                 pass
         return {}
+
+
+async def classify_task(task_id, iteration_id="", who=None, agent_id=None):
+    """判断任务来源：new_dev(新开发) / bug_fix(修复 Bug) / rework(质检重做)。
+
+    优先读任务 params.task_kind 标记（三条创建路径已打）；老任务无标记则按
+    params.rollback_from / title 兜底推断。develop 据此决定是否走 fix_bug 状态机：
+      - bug_fix / rework：任务开始 start_fix、完成 fix_bug
+      - new_dev：纯开发，不走 bug 状态机（除非过程中自己 report 了新 bug）
+
+    返回 (ok, {"task_kind": ..., "reason": ..., "title": ...} 或错误消息)。
+    """
+    if not task_id:
+        return False, "缺少 task_id"
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        recs = await sor.sqlExe(
+            "SELECT id, title, params FROM pipeline_tasks WHERE id=${tid}$", {"tid": task_id})
+        await sor.sqlExe("COMMIT", {})
+        if not recs:
+            return False, f"任务不存在: {task_id}"
+        rec = recs[0]
+        title = getattr(rec, 'title', '') or ''
+        params_str = getattr(rec, 'params', '{}') or '{}'
+        try:
+            params = json.loads(params_str) if isinstance(params_str, str) else (params_str or {})
+        except (json.JSONDecodeError, TypeError):
+            params = {}
+
+        kind = (params.get('task_kind') or '').strip()
+        reason = "params.task_kind 标记"
+        if not kind:
+            # 老任务兜底：rollback_from 存在 → rework；title 含「修复 Bug」→ bug_fix；否则 new_dev
+            if params.get('rollback_from'):
+                kind = 'rework'
+                reason = "params.rollback_from 存在（回退重做）"
+            elif ('修复 Bug' in title) or ('修复Bug' in title) or ('修复bug' in title):
+                kind = 'bug_fix'
+                reason = "title 含「修复 Bug」"
+            else:
+                kind = 'new_dev'
+                reason = "无 bug 相关标记，默认新开发"
+        return True, {"task_id": task_id, "task_kind": kind, "title": title, "reason": reason}

pipeline_service/capability_tools.py

@@ -139,6 +139,12 @@ TOOL_SCHEMAS = {
         "params": {"status": "状态(可选)", "severity": "严重度(可选)"},
         "required": [],
     },
+    "classify_task": {
+        "module": "bug_capability",
+        "description": "判断任务来源：new_dev(新开发)/bug_fix(修复Bug)/rework(质检重做)。develop 接任务后先调用，据此决定是否走 start_fix→fix_bug 状态机",
+        "params": {"task_id": "任务ID(可选，默认当前任务)"},
+        "required": [],
+    },
     "start_fix": {
         "module": "bug_capability",
         "description": "开始修复 Bug：open/confirmed → fixing（记录处理人）。回退重做/修复 Bug 任务里修完代码必须调用，open 状态也可直接修复（PM review_rollback 回退时不走 confirm_bug）",
@@ -269,6 +275,7 @@ async def exec_capability_tool(tool_name, params, ctx):
     p.setdefault("iteration_id", ctx.get("iteration_id", ""))
     p.setdefault("who", ctx.get("who", ""))
     p.setdefault("agent_id", ctx.get("agent_id", ""))
+    p.setdefault("task_id", ctx.get("task_id", ""))
 
     try:
         mod = importlib.import_module(f"pipeline_service.{schema['module']}")