v3.4.0: agents produce real code files + git push + PM clones repos

- All agents can write code files via 'files' array in output JSON
- Auto git commit+push after code file creation
- PM agent clones project repos on first review
- PM review shows repository state
- shell_exec now allows ~/pipeline_ws
- Role-specific prompts guide code/file output

pipeline_service/agent_loop.py

@@ -1,32 +1,23 @@
-"""Agent loop — 角色agent循环认领执行任务（v3.3.0）。
+"""Agent loop — 角色agent循环认领执行任务（v3.4.0）。
 
-v3.3.0 新增：
+v3.4.0 新增：
-- PM（项目经理）角色：审核交付件，决定通过/驳回
+- 所有agent具备文件读写 + git pull/push 能力
-- 任务链：requirement → design → develop → test → deploy
+- agent 产出实际代码文件到项目仓库，而非仅文档
-  前序角色完成后 → PM审核通过 → 自动创建下一角色任务
+- PM agent 负责 clone 项目关联仓库、初始化工作区
-- 产物文件输出：交付件写入项目工作目录 deliverables/{role}/
+- develop agent 产出可直接运行的源码文件
-- 工作目录自动创建：确保 workspace_dir 存在
+- 每个agent完成后自动 git commit + push
 
-架构：
+任务链：requirement → design → develop → test → deploy
-- 主agent（cockpit_chat）理解客户意图，开发类意图写入 pipeline_tasks（带 role）。
-- 各角色agent（需求/设计/开发/测试/运维）调 role_agent_run/role_agent_loop
-  循环认领自己角色的任务并执行。
-- PM agent 调 pm_review_loop 循环审核处于 review 状态的任务。
-- 认领是原子的：UPDATE ... WHERE state='submitted' + claimed_by 令牌校验，
-  多agent并发不会重复认领。
-- 角色agent缺信息时返回 need_info → 写入 pipeline_agent_questions，
-  任务置 waiting，等主agent/客户回答后回到 submitted 重新认领继续。
-- 只认领没有步骤记录的任务（带步骤的 DAG 任务归 executor 管）。
 """
 
 import asyncio
 import json
 import os
+import subprocess
 import logging
 
 logger = logging.getLogger("pipeline.agent_loop")
 
-# 角色别名归一化：前端/主agent传的角色名统一到认领用的规范名
 ROLE_ALIASES = {
     'developer': 'develop', 'dev': 'develop', 'development': 'develop', 'coding': 'develop',
     'requirement': 'requirement', 'requirements': 'requirement', 'requirement_analysis': 'requirement',
@@ -37,19 +28,24 @@ ROLE_ALIASES = {
     'pm': 'pm', 'project_manager': 'pm', '项目经理': 'pm', 'manager': 'pm',
 }
 
-# 任务链：前序角色完成后，PM审核通过 → 自动创建下一角色任务
-# None 表示链末端
 ROLE_CHAIN = {
     'requirement': 'design',
     'design': 'develop',
     'develop': 'test',
     'test': 'deploy',
-    'deploy': None,  # 部署完成 = 链结束
+    'deploy': None,
 }
 
-# PM 审核后任务状态
+TASK_REVIEW = 'review'
-TASK_REVIEW = 'review'    # 角色agent完成，待PM审核
+TASK_APPROVED = 'approved'
-TASK_APPROVED = 'approved'  # PM审核通过
+
+# 允许的 shell 工作目录前缀（安全限制）
+_ALLOWED_WORKDIRS = [
+    os.path.expanduser('~/pipeline_ws'),
+    '/d/pipeline/workspaces',
+    '/tmp/pipeline_workspaces',
+    '/tmp/pipeline_ws',
+]
 
 
 def _normalize_role(role):
@@ -57,66 +53,6 @@ def _normalize_role(role):
     return ROLE_ALIASES.get(r, r)
 
 
-ROLE_PROMPT = """你是软件开发产线项目中的「__ROLE__」角色agent。
-
-当前认领到的任务：
-标题：__TITLE__
-参数：__PARAMS__
-__QNA__
-
-## 项目信息
-- 工作目录：__WORKSPACE__
-- 你的产出物将保存到：__DELIVERABLE_DIR__
-
-请站在你的角色立场完成这项工作。
-
-## 产出规范
-1. 产出内容必须完整、可交付、可直接使用
-2. 产出格式：根据角色不同，产出对应格式：
-   - requirement（需求分析）：需求规格文档（Markdown格式）
-   - design（设计）：设计文档 / 架构图描述（Markdown格式）
-   - develop（开发）：可运行的代码 + 代码说明
-   - test（测试）：测试用例 + 测试报告（Markdown格式）
-   - deploy（部署）：部署文档 / 配置 / 脚本
-
-输出要求——返回纯 JSON（不要 markdown 包裹）：
-- 能完成时：{"status":"done","deliverable_type":"文档类型(requirement_doc/design_doc/code/test_report/deploy_doc)","result":"完整交付内容","summary":"一段话概述产出的关键结论"}
-- 缺少关键信息无法继续时：{"status":"need_info","question":"要向主agent/客户提出的具体问题","partial":"已完成的部分"}"""
-
-PM_PROMPT = """你是软件开发产线项目中的「项目经理（PM）」角色agent。
-
-你的职责是审核其他角色agent产出的交付件，确保质量和完整性，并推动项目前进。
-
-## 任务链
-项目按照以下阶段推进：
-1. requirement（需求分析）→ 2. design（设计）→ 3. develop（开发）→ 4. test（测试）→ 5. deploy（部署）
-
-当前待审核的任务：
-标题：__TITLE__
-角色：__ROLE__
-交付件内容：
-__DELIVERABLE__
-
-## 你的决策
-请基于交付件内容做出判断，输出纯JSON（不要markdown包裹）：
-
-1. 审核通过，启动下一阶段：
-   {"status":"approved","comment":"审核意见","next_task_title":"下一阶段任务标题","next_task_description":"下一阶段任务详细描述"}
-
-2. 审核不通过，需要修改：
-   {"status":"rejected","comment":"驳回原因和具体修改要求","questions":"需要角色agent澄清的具体问题"}
-
-3. 项目已完成（deploy阶段通过后）：
-   {"status":"completed","comment":"项目完成总结"}
-
-## 审核标准
-- requirement：需求是否清晰、完整、可量化
-- design：方案是否合理、技术选型是否恰当、是否覆盖需求
-- develop：代码是否可运行、是否遵循规范、是否实现设计
-- test：测试覆盖是否充分、是否发现关键问题
-- deploy：部署文档是否完整、配置是否正确"""
-
-
 def _get_db():
     from sqlor.dbpools import DBPools
     db = DBPools()
@@ -129,7 +65,6 @@ def _get_db():
 
 
 def _resolve_workspace(workspace_dir):
-    """解析工作目录，如果不可写则回退到 ~/pipeline_ws。确保目录存在。"""
     if not workspace_dir:
         workspace_dir = os.path.expanduser('~/pipeline_ws/default')
     try:
@@ -143,22 +78,186 @@ def _resolve_workspace(workspace_dir):
     return workspace_dir
 
 
-async def _claim_task(sor, tenant_id: str, role: str, state: str = 'submitted', match_role: bool = True):
+# ── Agent 工具函数 ──
-    """原子认领一条符合角色的任务。返回 task record 或 None。
 
-    认领策略：
+def _is_safe_workdir(workdir):
-    1. 选出最早一条符合状态和角色的任务，
+    """检查目录是否在允许范围内。"""
-       且排除有步骤记录的 DAG 任务（那些归 executor）。
+    wd = os.path.abspath(workdir)
-    2. UPDATE ... SET state='running', claimed_by=令牌 WHERE id=? AND state=原状态
+    for allowed in _ALLOWED_WORKDIRS:
-    3. 用令牌回查校验——别的agent抢先时校验失败，返回 None。
+        awd = os.path.abspath(os.path.expanduser(allowed))
+        if wd.startswith(awd):
+            return True
+    return False
 
-    match_role=False 时不过滤角色（PM审核所有角色的review任务）。
+
-    """
+async def _run_shell(command, workdir, timeout=120):
+    """安全执行 shell 命令。返回 {"rc": int, "stdout": str, "stderr": str}"""
+    cwd = os.path.abspath(workdir) if workdir else os.path.expanduser('~/pipeline_ws')
+    if not _is_safe_workdir(cwd):
+        return {"rc": -1, "stdout": "", "stderr": f"安全限制：目录 {cwd} 不在允许范围"}
+    if not os.path.isdir(cwd):
+        return {"rc": -1, "stdout": "", "stderr": f"目录不存在: {cwd}"}
+    try:
+        proc = await asyncio.create_subprocess_shell(
+            command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
+            cwd=cwd, executable='/bin/bash')
+        try:
+            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+        except asyncio.TimeoutError:
+            proc.kill()
+            await proc.wait()
+            return {"rc": -1, "stdout": "", "stderr": f"命令超时（{timeout}s）"}
+        return {"rc": proc.returncode or 0,
+                "stdout": stdout.decode('utf-8', 'replace')[-8000:],
+                "stderr": stderr.decode('utf-8', 'replace')[-4000:]}
+    except Exception as e:
+        return {"rc": -1, "stdout": "", "stderr": str(e)[:500]}
+
+
+async def _write_code_file(filepath, content):
+    """写代码文件，自动创建父目录。"""
+    try:
+        d = os.path.dirname(filepath)
+        if d:
+            os.makedirs(d, exist_ok=True)
+        with open(filepath, 'w', encoding='utf-8') as f:
+            f.write(content or '')
+        return True, filepath
+    except Exception as e:
+        return False, str(e)
+
+
+async def _git_setup(workdir):
+    """确保 git 用户已配置。"""
+    r = await _run_shell('git config user.email', workdir, 5)
+    if 'opencomputing' not in r.get('stdout', '') and '@' not in r.get('stdout', ''):
+        await _run_shell('git config user.email "pipeline@opencomputing.cn"', workdir, 5)
+        await _run_shell('git config user.name "Pipeline Agent"', workdir, 5)
+
+
+async def _git_commit_push(workdir, commit_message, branch='main'):
+    """git add → commit → push。"""
+    await _git_setup(workdir)
+    r1 = await _run_shell('git add -A', workdir, 10)
+    r2 = await _run_shell(f'git diff --cached --stat', workdir, 10)
+    if not r2.get('stdout', '').strip():
+        return {"rc": 0, "message": "没有变更需要提交"}
+    r3 = await _run_shell(f'git commit -m "{commit_message}"', workdir, 15)
+    if r3['rc'] != 0:
+        return {"rc": r3['rc'], "message": f"commit 失败: {r3['stderr'][:200]}"}
+    r4 = await _run_shell(f'git push origin {branch}', workdir, 30)
+    return {"rc": r4['rc'], "message": f"push {'成功' if r4['rc']==0 else '失败'}: {r4['stderr'][:200]}"}
+
+
+async def _git_clone(repo_url, target_dir, branch='main'):
+    """克隆仓库到目标目录。已存在则 pull。"""
+    if os.path.isdir(os.path.join(target_dir, '.git')):
+        r = await _run_shell(f'git checkout {branch} && git pull origin {branch}', target_dir, 30)
+        return {"rc": r['rc'], "message": f"已存在，pull: {r['stdout'][:200]}"}
+    parent = os.path.dirname(target_dir)
+    os.makedirs(parent, exist_ok=True)
+    r = await _run_shell(f'git clone -b {branch} {repo_url} {target_dir}', parent, 120)
+    return {"rc": r['rc'], "message": f"clone: {r['stdout'][:200] if r['rc']==0 else r['stderr'][:200]}"}
+
+
+# ── Prompts ──
+
+ROLE_PROMPT = """你是软件开发产线项目中的「__ROLE__」角色agent。
+
+当前任务：__TITLE__
+参数：__PARAMS__
+__QNA__
+
+## 工作环境
+- 项目根目录：__WORKSPACE__
+- 代码仓库位于：__WORKSPACE__/repos/ 下各子目录
+- 你有能力读写文件、执行git操作
+
+## 产出要求
+__ROLE_SPECIFIC__
+
+## 输出格式（严格JSON，不要markdown包裹）
+{
+  "status": "done",
+  "deliverable_type": "类型(requirement_doc/design_doc/code_files/test_report/deploy_doc)",
+  "summary": "一段话概述产出",
+  "result": "当产出是文档时，这里是完整文档内容",
+  "files": [
+    {"path": "repos/仓库名/src/相对路径/文件.java", "content": "文件完整内容"},
+    {"path": "repos/仓库名/pom.xml", "content": "..."}
+  ],
+  "git_commit_message": "feat: 简短描述本次变更",
+  "need_more_info": false,
+  "question": ""
+}
+
+- 如果是文档类产出(requirement/design)，用 result 字段输出文档正文
+- 如果是代码产出(develop)，必须用 files 数组输出每个源码文件，每个元素含 path 和 content
+- 测试产出用 result 输出测试报告，如有测试脚本则用 files
+- 部署产出用 files 输出部署配置/脚本，用 result 输出部署说明
+- 缺少关键信息时：{"status":"need_info","question":"问题"}
+
+## Git 操作说明
+- 产出代码文件将自动写入 files 中指定的路径并 git commit + push
+- 你只需要在 files 中指定正确的仓库路径即可"""
+
+ROLE_SPECIFICS = {
+    'requirement': """你是需求分析师。输出完整的需求规格文档（Markdown），用 result 字段。
+文档应包含：项目概述、功能需求、非功能需求、用户角色、业务流程、验收标准。""",
+
+    'design': """你是系统设计师。输出完整的设计文档（Markdown），用 result 字段。
+文档应包含：系统架构、技术选型、数据库设计(ER图描述)、API接口设计、前端页面规划、安全设计。
+同时可输出 proto/接口定义文件到 files。""",
+
+    'develop': """你是开发工程师。你必须输出实际可运行的代码文件！
+用 files 数组输出每个源码文件，准确指定在仓库中的路径。
+files 中的 path 格式：repos/{仓库名}/src/... 或 repos/{仓库名}/pom.xml 等。
+每个文件包含完整可运行的代码。
+同时用 result 字段输出代码说明。""",
+
+    'test': """你是测试工程师。输出测试报告（Markdown）到 result 字段。
+如有测试脚本/用例代码，输出到 files。
+报告应包含：测试策略、测试用例、执行结果、Bug清单、覆盖率。""",
+
+    'deploy': """你是部署运维工程师。输出部署文档（Markdown）到 result 字段。
+必须输出部署配置/脚本到 files：Dockerfile/docker-compose.yml/nginx配置等。
+files 中的 path 格式：repos/{仓库名}/deploy/...""",
+}
+
+PM_PROMPT = """你是项目经理（PM）。审核交付件并推动项目前进。
+
+## 项目
+工作目录：__WORKSPACE__
+关联仓库：__REPOS__
+
+## 待审核任务
+标题：__TITLE__
+角色：__ROLE__
+交付件摘要：__SUMMARY__
+交付件内容(前6000字)：__DELIVERABLE__
+
+## 仓库状态
+__REPO_STATE__
+
+## 决策
+输出纯JSON：
+1. 通过：{"status":"approved","comment":"审核意见","next_task_title":"下阶段标题","next_task_description":"详细描述"}
+2. 驳回：{"status":"rejected","comment":"驳回原因","questions":"修改要求"}
+3. 完成：{"status":"completed","comment":"项目总结"}
+
+审核标准：
+- requirement：需求是否清晰完整可量化
+- design：方案合理、覆盖需求、技术可行
+- develop：代码文件是否实际产出、目录结构是否合理、是否可编译运行
+- test：测试覆盖充分、发现问题记录完整
+- deploy：部署配置完整、可一键部署"""
+
+
+# ── Agent 核心逻辑 ──
+
+async def _claim_task(sor, tenant_id, role, state='submitted', match_role=True):
     role = _normalize_role(role)
-    if match_role:
+    where_role = "AND (role=${role}$ OR role='')" if match_role else ""
-        where_role = "AND (role=${role}$ OR role='')"
-    else:
-        where_role = ""
     recs = await sor.sqlExe(
         "SELECT id, title, params, pipeline_id, tenant_id, role FROM pipeline_tasks "
         "WHERE tenant_id=${tid}$ AND state=${state}$ " + where_role + " "
@@ -167,53 +266,43 @@ async def _claim_task(sor, tenant_id: str, role: str, state: str = 'submitted',
         {"tid": tenant_id, "state": state, "role": role})
     if not recs:
         return None
-
     task = recs[0]
     task_id = task.id
-
     from appPublic.uniqueID import getID
     claim_token = getID()
     await sor.sqlExe(
         "UPDATE pipeline_tasks SET state='running', claimed_by=${cb}$ "
         "WHERE id=${tid}$ AND state=${state}$",
         {"cb": claim_token, "tid": task_id, "state": state})
-
-    # 令牌校验：确认是本agent抢到的
     check = await sor.sqlExe(
         "SELECT id FROM pipeline_tasks WHERE id=${tid}$ AND state='running' AND claimed_by=${cb}$",
         {"tid": task_id, "cb": claim_token})
     if not check:
-        logger.info(f"claim lost race: task={task_id} role={role}")
+        logger.info(f"claim lost race: task={task_id}")
         return None
     return task
 
 
 async def _get_workspace_dir(sor, project_id):
-    """获取项目的工作目录，确保可写并存在。"""
+    recs = await sor.sqlExe("SELECT workspace_dir FROM sd_projects WHERE id=${pid}$", {"pid": project_id})
-    recs = await sor.sqlExe(
-        "SELECT workspace_dir FROM sd_projects WHERE id=${pid}$",
-        {"pid": project_id})
     ws = getattr(recs[0], 'workspace_dir', '') if recs else ''
     return _resolve_workspace(ws)
 
 
-async def _build_qna_section(sor, task_id: str) -> str:
+async def _build_qna_section(sor, task_id):
-    """把该任务已回答的问答历史注入prompt，让角色agent带着答案继续。"""
     from .questions import get_task_qna
     qna = await get_task_qna(task_id)
     if not qna:
         return ""
-    lines = ["历史问答（这些问题已得到回答，请结合答案继续工作）："]
+    lines = ["历史问答："]
     for item in qna:
-        src = item.get('answer_source') or ''
+        src = "客户" if item.get('answer_source') == "customer" else "主agent"
-        src_label = "客户" if src == "customer" else "主agent"
+        lines.append(f"问：{item.get('question','')}")
-        lines.append(f"问：{item.get('question', '')}")
+        lines.append(f"答({src})：{item.get('answer','')}")
-        lines.append(f"答（{src_label}）：{item.get('answer', '')}")
     return "\n".join(lines)
 
 
-def _parse_result(raw: str) -> dict:
+def _parse_result(raw):
-    """解析角色agent的 LLM 返回。解析失败时按 done + 原文处理。"""
     raw = (raw or "").strip()
     if raw.startswith("```"):
         raw = raw.split("\n", 1)[1].rsplit("```", 1)[0]
@@ -226,42 +315,58 @@ def _parse_result(raw: str) -> dict:
     return {"status": "done", "result": raw}
 
 
-async def _write_deliverable_file(workspace_dir, role, task_id, deliverable_type, content):
-    """将交付件写入项目工作目录的文件中。"""
-    role_dir = os.path.join(workspace_dir, 'deliverables', role)
-    os.makedirs(role_dir, exist_ok=True)
-    # 文件名：任务ID_类型.md
-    safe_type = (deliverable_type or 'deliverable').replace('/', '_')
-    filename = f"{task_id}_{safe_type}.md"
-    file_path = os.path.join(role_dir, filename)
-    try:
-        with open(file_path, 'w', encoding='utf-8') as f:
-            f.write(content or '')
-        logger.info(f"deliverable written: {file_path}")
-        return file_path
-    except Exception as e:
-        logger.error(f"failed to write deliverable file: {file_path} err={e}")
-        return ''
-
-
 async def _get_deliverable_content(sor, task_id):
-    """获取任务的最新交付件内容。"""
     recs = await sor.sqlExe(
         "SELECT content, deliverable_type FROM pipeline_deliverables "
-        "WHERE task_id=${tid}$ ORDER BY created_at DESC LIMIT 1",
+        "WHERE task_id=${tid}$ ORDER BY created_at DESC LIMIT 1", {"tid": task_id})
-        {"tid": task_id})
     if recs:
         return getattr(recs[0], 'content', '') or '', getattr(recs[0], 'deliverable_type', '') or ''
     return '', ''
 
 
+async def _get_project_repos(sor, project_id):
+    """获取项目关联的所有仓库。"""
+    recs = await sor.sqlExe(
+        "SELECT id, repo_name, repo_url, default_branch, local_path FROM sd_project_repos "
+        "WHERE project_id=${pid}$", {"pid": project_id})
+    return [{'id': getattr(r, 'id', ''), 'name': getattr(r, 'repo_name', ''),
+             'url': getattr(r, 'repo_url', ''), 'branch': getattr(r, 'default_branch', 'main'),
+             'local_path': getattr(r, 'local_path', '') or ''} for r in (recs or [])]
+
+
+async def _setup_repos(sor, workspace_dir, project_id):
+    """PM：clone 所有项目关联仓库到 workspace/repos/。"""
+    repos = await _get_project_repos(sor, project_id)
+    results = []
+    repos_dir = os.path.join(workspace_dir, 'repos')
+    os.makedirs(repos_dir, exist_ok=True)
+    for repo in repos:
+        target = os.path.join(repos_dir, repo['name'])
+        r = await _git_clone(repo['url'], target, repo['branch'])
+        results.append({'repo': repo['name'], **r})
+    return results
+
+
+async def _get_repo_state(workspace_dir):
+    """获取仓库当前状态（供PM审核时查看）。"""
+    repos_dir = os.path.join(workspace_dir, 'repos')
+    if not os.path.isdir(repos_dir):
+        return "暂无仓库"
+    lines = []
+    for name in sorted(os.listdir(repos_dir)):
+        rp = os.path.join(repos_dir, name)
+        if os.path.isdir(rp) and os.path.isdir(os.path.join(rp, '.git')):
+            r = await _run_shell('git log --oneline -3', rp, 5)
+            lines.append(f"\n[{name}]")
+            lines.append(r.get('stdout', '') or '(空仓库)')
+    return "\n".join(lines) if lines else "仓库无提交记录"
+
+
 async def _get_next_role(current_role):
-    """获取任务链中的下一个角色。"""
     return ROLE_CHAIN.get(_normalize_role(current_role))
 
 
 async def _create_next_task(sor, project_id, task, next_role, pm_comment=''):
-    """为下一个角色创建任务。"""
     from appPublic.uniqueID import getID
     title = getattr(task, 'title', '') or ''
     params_str = getattr(task, 'params', '{}') or '{}'
@@ -269,47 +374,26 @@ async def _create_next_task(sor, project_id, task, next_role, pm_comment=''):
         params = json.loads(params_str) if isinstance(params_str, str) else params_str
     except (json.JSONDecodeError, TypeError):
         params = {}
-
     new_title = f"{title}（{next_role}阶段）"
-    new_params = {
+    new_params = {**params, 'previous_role': _normalize_role(getattr(task, 'role', '')),
-        **params,
+                  'previous_task_id': getattr(task, 'id', ''), 'pm_comment': pm_comment}
-        'previous_role': _normalize_role(getattr(task, 'role', '')),
-        'previous_task_id': getattr(task, 'id', ''),
-        'pm_comment': pm_comment,
-    }
-
     new_task_id = getID()
     await sor.C('pipeline_tasks', {
-        'id': new_task_id,
+        'id': new_task_id, 'tenant_id': project_id, 'pipeline_id': 'role_task',
-        'tenant_id': project_id,
+        'owner_id': 'pm', 'title': new_title,
-        'pipeline_id': 'role_task',
-        'owner_id': 'pm',
-        'title': new_title,
         'params': json.dumps(new_params, ensure_ascii=False),
-        'role': next_role,
+        'role': next_role, 'state': 'submitted', 'claimed_by': None,
-        'state': 'submitted',
-        'claimed_by': None,
     })
-    logger.info(f"_create_next_task: {next_role} task={new_task_id} from {getattr(task, 'id', '')}")
+    logger.info(f"_create_next_task: {next_role} task={new_task_id}")
     return new_task_id, new_title
 
 
-async def role_agent_run(project_id: str, role: str, agent_id: str = None, model_name: str = None) -> dict:
+# ── 角色 Agent ──
-    """角色agent单次迭代：认领一条任务 → 执行 → 交付。
 
-    非PM角色完成 → 任务置 review（等PM审核）
+async def role_agent_run(project_id, role, agent_id=None, model_name=None):
-    PM角色 → 由 pm_review_run 处理
-
-    Returns:
-        {"status": "idle"} 没有待办任务
-        {"status": "completed", "task_id", "deliverable_id"} 完成交付（已进入review）
-        {"status": "need_info", "task_id", "question_id", "question"} 缺信息已提问
-        {"status": "failed", "task_id", "error"} 执行失败
-    """
     role = _normalize_role(role)
     if role == 'pm':
-        # PM不执行角色任务，走pm_review_run
+        return {"status": "idle", "message": "PM请使用 pm_review_run"}
-        return {"status": "idle", "message": "PM agent请使用 pm_review_run"}
 
     db = _get_db()
     async with db.sqlorContext("pipeline") as sor:
@@ -320,74 +404,198 @@ async def role_agent_run(project_id: str, role: str, agent_id: str = None, model
         task_id = task.id
         title = getattr(task, "title", "") or ""
         params_str = getattr(task, "params", "{}") or "{}"
-
-        # 获取工作目录
         workspace_dir = await _get_workspace_dir(sor, project_id)
-        deliverable_dir = os.path.join(workspace_dir, 'deliverables', role)
-        os.makedirs(deliverable_dir, exist_ok=True)
 
+        # 确保 repos 目录存在
+        repos_dir = os.path.join(workspace_dir, 'repos')
+        if not os.path.isdir(repos_dir):
+            os.makedirs(repos_dir, exist_ok=True)
+
+        role_specific = ROLE_SPECIFICS.get(role, ROLE_SPECIFICS.get('develop', ''))
         qna_section = await _build_qna_section(sor, task_id)
+
         prompt = (ROLE_PROMPT
                   .replace('__ROLE__', role)
                   .replace('__TITLE__', title)
                   .replace('__PARAMS__', params_str)
                   .replace('__QNA__', qna_section)
                   .replace('__WORKSPACE__', workspace_dir)
-                  .replace('__DELIVERABLE_DIR__', deliverable_dir))
+                  .replace('__ROLE_SPECIFIC__', role_specific))
 
         from .llm_bridge import llm_call
         try:
             raw = await llm_call(prompt, model=model_name, temperature=0.4)
         except Exception as e:
-            await sor.sqlExe(
+            await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id})
-                "UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$",
-                {"tid": task_id})
             logger.error(f"role_agent_run llm failed: task={task_id} err={e}")
             return {"status": "failed", "task_id": task_id, "error": str(e)[:200]}
 
         parsed = _parse_result(raw)
 
-        # 缺信息 → 提问回路：任务置 waiting，等主agent/客户回答
         if parsed.get("status") == "need_info" and parsed.get("question"):
             from .questions import agent_ask
             qid = await agent_ask(project_id, task_id, role, parsed["question"],
                                   context={"partial": parsed.get("partial", ""), "title": title})
-            return {"status": "need_info", "task_id": task_id,
+            return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": parsed["question"]}
-                    "question_id": qid, "question": parsed["question"]}
 
-        # 完成 → 写交付件到文件 + DB
+        # ── 处理产出 ──
         from appPublic.uniqueID import getID
         did = getID()
         result_text = parsed.get("result") or raw
         deliverable_type = parsed.get("deliverable_type") or role
         summary = parsed.get("summary", "")
 
-        # 写文件
+        # 写入实际代码文件（如果有 files 数组）
-        file_path = await _write_deliverable_file(
+        files_written = []
-            workspace_dir, role, task_id, deliverable_type, result_text)
+        code_files = parsed.get("files") or []
+        if isinstance(code_files, list):
+            for f in code_files:
+                if isinstance(f, dict) and f.get("path") and f.get("content"):
+                    abs_path = os.path.join(workspace_dir, f["path"])
+                    ok, msg = await _write_code_file(abs_path, f["content"])
+                    if ok:
+                        files_written.append(abs_path)
+                        logger.info(f"code file written: {abs_path}")
+                    else:
+                        logger.error(f"code file failed: {abs_path} err={msg}")
+
+        # 写交付件文档
+        file_path = ''
+        if result_text:
+            deliverable_dir = os.path.join(workspace_dir, 'deliverables', role)
+            os.makedirs(deliverable_dir, exist_ok=True)
+            safe_type = (deliverable_type or 'deliverable').replace('/', '_')
+            file_path = os.path.join(deliverable_dir, f"{task_id}_{safe_type}.md")
+            try:
+                with open(file_path, 'w', encoding='utf-8') as f:
+                    f.write(result_text or '')
+                logger.info(f"deliverable written: {file_path}")
+            except Exception as e:
+                logger.error(f"deliverable write failed: {file_path} err={e}")
+                file_path = ''
 
         # 写DB
         await sor.C("pipeline_deliverables", {
             "id": did, "project_id": project_id, "task_id": task_id,
-            "deliverable_type": deliverable_type,
+            "deliverable_type": deliverable_type, "title": title,
-            "title": title, "content": result_text,
+            "content": result_text, "file_path": file_path,
-            "file_path": file_path,
             "quality_score": 80, "review_status": "pending",
             "created_by": agent_id or role,
         })
 
-        # 置为 review 状态，等待PM审核（清除 claimed_by 让 PM poller 能认领）
+        # Git commit + push（如果有产出文件）
+        commit_msg = parsed.get("git_commit_message") or f"{role}: {title[:80]}"
+        git_result = {"rc": 0, "message": "无git操作"}
+        if files_written:
+            # 在 repos 目录下的每个 git 仓库各自 commit
+            for repo_name in os.listdir(repos_dir):
+                repo_path = os.path.join(repos_dir, repo_name)
+                if os.path.isdir(os.path.join(repo_path, '.git')):
+                    git_result = await _git_commit_push(repo_path, commit_msg)
+
+        # 置为 review 状态
         await sor.sqlExe(
             "UPDATE pipeline_tasks SET state=${st}$, claimed_by=NULL WHERE id=${tid}$",
             {"st": TASK_REVIEW, "tid": task_id})
-        logger.info(f"role_agent_run completed(review): task={task_id} role={role} deliverable={did} file={file_path}")
+        logger.info(f"role_agent_run completed: task={task_id} role={role} "
+                     f"deliverable={did} files={len(files_written)} git={git_result.get('rc',-1)}")
         return {"status": "completed", "task_id": task_id, "deliverable_id": did,
-                "next_state": TASK_REVIEW, "file_path": file_path}
+                "files_written": len(files_written), "git_result": git_result}
 
 
-async def role_agent_loop(project_id: str, role: str, agent_id: str = None,
+# ── PM 审核 ──
-                          model_name: str = None, max_iterations: int = 10) -> list:
+
-    """角色agent循环：连续认领执行，直到没有任务、需要提问或达到上限。"""
+async def pm_review_run(project_id, agent_id=None, model_name=None):
+    db = _get_db()
+    async with db.sqlorContext("pipeline") as sor:
+        task = await _claim_task(sor, project_id, '', state=TASK_REVIEW, match_role=False)
+        if not task:
+            return {"status": "idle", "message": "没有待审核任务"}
+
+        task_id = task.id
+        title = getattr(task, "title", "") or ""
+        task_role = _normalize_role(getattr(task, "role", "") or "")
+        workspace_dir = await _get_workspace_dir(sor, project_id)
+
+        # 首次审核时，确保仓库已 clone
+        repos = await _get_project_repos(sor, project_id)
+        if repos:
+            await _setup_repos(sor, workspace_dir, project_id)
+
+        deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id)
+        if not deliverable_content:
+            await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", {"tid": task_id})
+            return {"status": "rejected", "task_id": task_id, "reason": "没有交付件"}
+
+        summary = deliverable_content[:200].replace('\n', ' ')
+        content_preview = deliverable_content[:6000]
+        if len(deliverable_content) > 6000:
+            content_preview += "\n...（截断）"
+
+        repo_state = await _get_repo_state(workspace_dir)
+        repos_str = ", ".join([r['name'] for r in repos]) if repos else "无"
+
+        prompt = (PM_PROMPT
+                  .replace('__TITLE__', title)
+                  .replace('__ROLE__', task_role)
+                  .replace('__SUMMARY__', summary)
+                  .replace('__DELIVERABLE__', content_preview)
+                  .replace('__WORKSPACE__', workspace_dir)
+                  .replace('__REPOS__', repos_str)
+                  .replace('__REPO_STATE__', repo_state))
+
+        from .llm_bridge import llm_call
+        try:
+            raw = await llm_call(prompt, model=model_name, temperature=0.3)
+        except Exception as e:
+            await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id})
+            return {"status": "failed", "task_id": task_id, "error": str(e)[:200]}
+
+        parsed = _parse_result(raw)
+        decision = parsed.get("status", "rejected")
+        comment = parsed.get("comment", "")
+
+        if decision == "approved":
+            from appPublic.uniqueID import getID
+            pm_did = getID()
+            await sor.C("pipeline_deliverables", {
+                "id": pm_did, "project_id": project_id, "task_id": task_id,
+                "deliverable_type": "pm_review",
+                "title": f"PM审核：{title}",
+                "content": json.dumps(parsed, ensure_ascii=False),
+                "file_path": os.path.join(workspace_dir, 'deliverables', 'pm', f"{task_id}_review.md"),
+                "quality_score": 100, "review_status": "approved",
+                "created_by": agent_id or "pm",
+            })
+            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}$ WHERE id=${tid}$",
+                {"st": TASK_APPROVED, "tid": task_id})
+
+            next_role = await _get_next_role(task_role)
+            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}
+            else:
+                return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"}
+
+        elif decision == "rejected":
+            from .questions import agent_ask
+            rejection_q = parsed.get("questions") or comment or "交付件不满足要求"
+            await agent_ask(project_id, task_id, "pm", rejection_q,
+                           context={"pm_comment": comment, "deliverable_type": deliverable_type})
+            await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", {"tid": task_id})
+            return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q}
+
+        else:
+            await sor.sqlExe("UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$", {"tid": task_id})
+            return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"}
+
+
+async def role_agent_loop(project_id, role, agent_id=None, model_name=None, max_iterations=10):
     results = []
     for _ in range(max_iterations):
         r = await role_agent_run(project_id, role, agent_id, model_name)
@@ -398,133 +606,7 @@ async def role_agent_loop(project_id: str, role: str, agent_id: str = None,
     return results
 
 
-# ── PM 审核 ──
+async def pm_review_loop(project_id, agent_id=None, model_name=None, max_iterations=20):
-
-async def pm_review_run(project_id: str, agent_id: str = None, model_name: str = None) -> dict:
-    """PM审核单次迭代：认领一条 review 状态的任务 → 审核交付件 → 批准/驳回。
-
-    Returns:
-        {"status": "idle"} 没有待审核任务
-        {"status": "approved", "task_id", "next_task_id", "next_role"} 审核通过，已创建下一阶段任务
-        {"status": "rejected", "task_id"} 驳回，任务回到 submitted
-        {"status": "completed", "task_id"} 项目完成（链末端）
-    """
-    db = _get_db()
-    async with db.sqlorContext("pipeline") as sor:
-        # 认领 review 状态的任务（不限角色，PM审核所有角色产出）
-        task = await _claim_task(sor, project_id, '', state=TASK_REVIEW, match_role=False)
-        if not task:
-            return {"status": "idle", "message": "没有待审核任务"}
-
-        task_id = task.id
-        title = getattr(task, "title", "") or ""
-        task_role = _normalize_role(getattr(task, "role", "") or "")
-
-        # 获取交付件内容
-        deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id)
-        if not deliverable_content:
-            # 没有交付件 → 驳回
-            await sor.sqlExe(
-                "UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$",
-                {"tid": task_id})
-            return {"status": "rejected", "task_id": task_id,
-                    "reason": "没有找到交付件内容，任务已退回"}
-
-        # 截断过长内容（LLM有token上限）
-        content_preview = deliverable_content[:6000]
-        if len(deliverable_content) > 6000:
-            content_preview += "\n\n...（内容过长已截断）"
-
-        prompt = (PM_PROMPT
-                  .replace('__TITLE__', title)
-                  .replace('__ROLE__', task_role)
-                  .replace('__DELIVERABLE__', content_preview))
-
-        from .llm_bridge import llm_call
-        try:
-            raw = await llm_call(prompt, model=model_name, temperature=0.3)
-        except Exception as e:
-            logger.error(f"pm_review_run llm failed: task={task_id} err={e}")
-            await sor.sqlExe(
-                "UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$",
-                {"tid": task_id})
-            return {"status": "failed", "task_id": task_id, "error": str(e)[:200]}
-
-        parsed = _parse_result(raw)
-        decision = parsed.get("status", "rejected")
-        comment = parsed.get("comment", "")
-
-        if decision == "approved":
-            # 获取工作目录
-            workspace_dir = await _get_workspace_dir(sor, project_id)
-
-            # 写PM审核记录到交付件
-            from appPublic.uniqueID import getID
-            pm_did = getID()
-            await sor.C("pipeline_deliverables", {
-                "id": pm_did, "project_id": project_id, "task_id": task_id,
-                "deliverable_type": "pm_review",
-                "title": f"PM审核：{title}",
-                "content": json.dumps(parsed, ensure_ascii=False),
-                "file_path": os.path.join(workspace_dir, 'deliverables', 'pm',
-                                         f"{task_id}_review.md"),
-                "quality_score": 100, "review_status": "approved",
-                "created_by": agent_id or "pm",
-            })
-
-            # 更新原交付件审核状态
-            await sor.sqlExe(
-                "UPDATE pipeline_deliverables SET review_status='approved', "
-                "review_comment=${cm}$ WHERE task_id=${tid}$",
-                {"cm": comment, "tid": task_id})
-
-            # 任务置为 approved
-            await sor.sqlExe(
-                "UPDATE pipeline_tasks SET state=${st}$ WHERE id=${tid}$",
-                {"st": TASK_APPROVED, "tid": task_id})
-
-            # 检查是否有下一阶段
-            next_role = await _get_next_role(task_role)
-            if next_role:
-                next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment)
-                logger.info(f"pm_review: approved task={task_id}, created next={next_tid} role={next_role}")
-                return {"status": "approved", "task_id": task_id,
-                        "next_task_id": next_tid, "next_role": next_role,
-                        "next_title": next_title, "comment": comment}
-            else:
-                # 链末端（deploy审核通过 = 项目完成）
-                logger.info(f"pm_review: project completed, task={task_id} role={task_role}")
-                return {"status": "completed", "task_id": task_id,
-                        "comment": comment or "项目所有阶段已完成"}
-
-        elif decision == "rejected":
-            # 驳回：任务回到 submitted，附带PM驳回原因作为question
-            from .questions import agent_ask
-            rejection_question = parsed.get("questions") or comment or "交付件不满足要求，请修改后重新提交"
-            await agent_ask(project_id, task_id, "pm", rejection_question,
-                           context={"pm_comment": comment, "deliverable_type": deliverable_type})
-            # 任务回到 submitted（通过回答问题的resume机制）
-            # 但agent_ask已经把任务置waiting，需要让主agent/客户确认后才能回到submitted
-            # 简化为：直接回 submitted
-            await sor.sqlExe(
-                "UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$",
-                {"tid": task_id})
-            logger.info(f"pm_review: rejected task={task_id}")
-            return {"status": "rejected", "task_id": task_id,
-                    "comment": comment, "question": rejection_question}
-
-        else:
-            # completed（项目完成）
-            await sor.sqlExe(
-                "UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$",
-                {"tid": task_id})
-            return {"status": "completed", "task_id": task_id,
-                    "comment": comment or "项目完成"}
-
-
-async def pm_review_loop(project_id: str, agent_id: str = None,
-                         model_name: str = None, max_iterations: int = 20) -> list:
-    """PM审核循环：连续审核处于 review 状态的任务。"""
     results = []
     for _ in range(max_iterations):
         r = await pm_review_run(project_id, agent_id, model_name)
@@ -535,10 +617,8 @@ async def pm_review_loop(project_id: str, agent_id: str = None,
     return results
 
 
-# ── 向后兼容别名（旧前端按 run_agent_loop/agent_loop 调用）──
+# 向后兼容
-
 async def run_agent_loop(project_id, role_name, model_name=None):
-    """旧接口：单次迭代。role_name 为空时按 'dev' 处理。"""
     role = _normalize_role(role_name or "develop")
     if role == 'pm':
         return await pm_review_run(project_id, model_name=model_name)
@@ -546,9 +626,7 @@ async def run_agent_loop(project_id, role_name, model_name=None):
 
 
 async def agent_loop(project_id, role_name, model_name=None, max_iterations=10):
-    """旧接口：循环执行。role_name 为空时按 'dev' 处理。"""
     role = _normalize_role(role_name or "develop")
     if role == 'pm':
         return await pm_review_loop(project_id, model_name=model_name, max_iterations=max_iterations)
-    return await role_agent_loop(project_id, role_name or "develop",
+    return await role_agent_loop(project_id, role_name or "develop", model_name=model_name, max_iterations=max_iterations)
-                                 model_name=model_name, max_iterations=max_iterations)

pipeline_service/init.py

@@ -41,7 +41,7 @@ from .agent_loop import role_agent_run, role_agent_loop, agent_loop, run_agent_l
 from .agent_loop import pm_review_run, pm_review_loop
 
 MODULE_NAME = "pipeline_service"
-MODULE_VERSION = "3.3.0"
+MODULE_VERSION = "3.4.0"
 
 
 async def pipeline_submit(tenant_id, pipeline_id, owner_id, title, params=None):
@@ -378,6 +378,7 @@ def pipeline_unregister_step_type(step_type):
 
 _SHELL_BASE_DIR = '/d/pipeline/workspaces'
 _WORKDIR_FALLBACK = '/tmp/pipeline_workspaces'
+_PIPELINE_WS = os.path.expanduser('~/pipeline_ws')
 
 def _resolve_workdir():
     """选择可写的工作目录根。"""
@@ -401,8 +402,9 @@ async def shell_exec(command: str, workdir: str = None, timeout: int = 60):
         cwd = os.path.abspath(cwd)
     base = os.path.abspath(_SHELL_BASE_DIR)
     fallback = os.path.abspath(_WORKDIR_FALLBACK)
-    if not (cwd.startswith(base) or cwd.startswith(fallback)):
+    pipeline_ws = os.path.abspath(_PIPELINE_WS)
-        return {"rc": -1, "stdout": "", "stderr": f"安全限制：工作目录必须在 {_SHELL_BASE_DIR} 下"}
+    if not (cwd.startswith(base) or cwd.startswith(fallback) or cwd.startswith(pipeline_ws)):
+        return {"rc": -1, "stdout": "", "stderr": f"安全限制：工作目录必须在允许范围内"}
     try:
         proc = await asyncio.create_subprocess_shell(
             command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,