diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 77124da..8f528e3 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -804,80 +804,116 @@ async def _get_project_name(sor, project_id): return '' -async def _ensure_project_repo(workspace_dir, project_name=''): - """确保项目过程仓库存在并 git init（阶段文档/QC审计/项目管理文档放这里）。 - 仓库名 = {项目名}_pc（见 project-directory-spec 规范），查不到项目名时回退 PROJECT_REPO_NAME。 +async def _get_project_dir(sor, project_id): + """返回项目目录 {space}/projects/{项目名}/（项目过程仓库 + docs/ + env/ + deliverables/ 所在）。 + + 新结构：项目目录从 {space}/{项目名} 迁到 {space}/projects/{项目名}/，与 apps/、modules/ 平级。 + """ + space_dir = await _get_space_dir(sor, project_id) + if not space_dir: + return '' + project_name = await _get_project_name(sor, project_id) + if not project_name: + return '' + return os.path.join(space_dir, 'projects', project_name) + + +async def _ensure_project_repo(project_dir): + """确保项目目录（项目过程仓库）存在并 git init（docs/ + env/ + spec.json + deliverables/ 放这里）。 + + 新结构：项目目录 projects/{项目名}/ 本身就是项目过程仓库（不再是 repos/{项目名}_pc/）。 幂等：已存在则跳过。项目过程仓库是本地仓库，无远程，只 commit 不 push。""" - repos_dir = os.path.join(workspace_dir, 'repos') - os.makedirs(repos_dir, exist_ok=True) - repo_name = f"{project_name}_pc" if project_name else PROJECT_REPO_NAME - repo_dir = os.path.join(repos_dir, repo_name) - os.makedirs(repo_dir, exist_ok=True) - if os.path.isdir(os.path.join(repo_dir, '.git')): + os.makedirs(project_dir, exist_ok=True) + if os.path.isdir(os.path.join(project_dir, '.git')): return {'rc': 0, 'message': '项目过程仓库已存在'} - async with _git_lock(repo_dir): - if os.path.isdir(os.path.join(repo_dir, '.git')): + async with _git_lock(project_dir): + if os.path.isdir(os.path.join(project_dir, '.git')): return {'rc': 0, 'message': '项目过程仓库已存在'} - r = await _run_shell('git init', repo_dir, 10) + r = await _run_shell('git init', project_dir, 10) if r['rc'] != 0: return {'rc': r['rc'], 'message': f"git init 失败: {r['stderr'][:200]}"} # init 之后才能配置仓库级 user（否则 git config 报 not a git repository） - await _git_setup(repo_dir) - readme = os.path.join(repo_dir, 'README.md') + await _git_setup(project_dir) + readme = os.path.join(project_dir, 'README.md') if not os.path.isfile(readme): try: with open(readme, 'w', encoding='utf-8') as f: f.write('# 项目过程仓库\n\n阶段文档 / QC 审计文档 / 项目管理文档。\n') except Exception: pass - r2 = await _run_shell('git add -A && git commit -m "init: 项目过程仓库"', repo_dir, 15) + r2 = await _run_shell('git add -A && git commit -m "init: 项目过程仓库"', project_dir, 15) return {'rc': r2['rc'], 'message': f"init 项目过程仓库 {'成功' if r2['rc'] == 0 else '失败'}"} -async def _commit_repos_after_approve(workspace_dir, title=''): - """审核通过后统一提交：项目过程仓库 repos/{项目名}_pc/（阶段/QC/PM 文档）+ repos/ 下应用/模块仓库（代码）。 +async def _commit_repos_after_approve(space_dir, project_name='', title=''): + """审核通过后统一提交：项目过程仓库 projects/{项目名}/（阶段/QC/PM 文档）+ apps/* + modules/* 仓库（代码）。 不在 agent 每次产出时提交，减少 git 并发锁与远端 push 频率。""" msg = f"approve: {title[:80]}" if title else "approve: 审核通过" - repos_dir = os.path.join(workspace_dir, 'repos') results = [] - if os.path.isdir(repos_dir): - for name in sorted(os.listdir(repos_dir)): - rp = os.path.join(repos_dir, name) + roots = [] + if project_name: + roots.append(('projects', os.path.join(space_dir, 'projects', project_name))) + roots.append(('apps', os.path.join(space_dir, 'apps'))) + roots.append(('modules', os.path.join(space_dir, 'modules'))) + for kind, base in roots: + if not os.path.isdir(base): + continue + if kind == 'projects': + if os.path.isdir(os.path.join(base, '.git')): + r = await _git_commit_push(base, msg) + results.append(f"{os.path.basename(base)}: {r.get('message', '')}") + logger.info(f"commit-after-approve projects: rc={r.get('rc', -1)} {r.get('message', '')[:100]}") + continue + for name in sorted(os.listdir(base)): + rp = os.path.join(base, name) if os.path.isdir(os.path.join(rp, '.git')): r = await _git_commit_push(rp, msg) - results.append(f"{name}: {r.get('message', '')}") - logger.info(f"commit-after-approve {name}: rc={r.get('rc', -1)} {r.get('message', '')[:100]}") + results.append(f"{kind}/{name}: {r.get('message', '')}") + logger.info(f"commit-after-approve {kind}/{name}: rc={r.get('rc', -1)} {r.get('message', '')[:100]}") return {"rc": 0, "message": "; ".join(results) if results else "无仓库可提交"} -async def _setup_repos(sor, workspace_dir, project_id): - """PM：clone 所有项目关联仓库到 workspace/repos/。""" - # 先确保项目过程仓库存在（阶段文档/QC审计/PM文档放这里），仓库名 = {项目名}_pc - project_name = await _get_project_name(sor, project_id) - await _ensure_project_repo(workspace_dir, project_name) +async def _setup_repos(sor, space_dir, project_id): + """clone 项目关联仓库到机构工作空间：应用→apps/、模块→modules/；项目目录→projects/{项目名}/。""" + project_dir = await _get_project_dir(sor, project_id) + await _ensure_project_repo(project_dir) 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']) + name = repo['name'] or '' + # 应用仓库：名字以 _app 结尾 → apps/{去掉_app}/；否则模块仓库 → modules/{name}/ + if name.endswith('_app'): + target = os.path.join(space_dir, 'apps', name[:-4]) + else: + target = os.path.join(space_dir, 'modules', name) r = await _git_clone(repo['url'], target, repo['branch']) - results.append({'repo': repo['name'], **r}) + results.append({'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 "暂无仓库" +async def _get_repo_state(space_dir, project_name=''): + """获取仓库当前状态（供PM审核时查看）：项目过程仓库 projects/{项目名}/ + apps/* + modules/*。""" 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 '(空仓库)') + roots = [] + if project_name: + roots.append(('projects', os.path.join(space_dir, 'projects', project_name))) + roots.append(('apps', os.path.join(space_dir, 'apps'))) + roots.append(('modules', os.path.join(space_dir, 'modules'))) + for kind, base in roots: + if not os.path.isdir(base): + continue + if kind == 'projects': + if os.path.isdir(os.path.join(base, '.git')): + r = await _run_shell('git log --oneline -3', base, 5) + lines.append(f"\n[{os.path.basename(base)}]") + lines.append(r.get('stdout', '') or '(空仓库)') + continue + for name in sorted(os.listdir(base)): + rp = os.path.join(base, 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[{kind}/{name}]") + lines.append(r.get('stdout', '') or '(空仓库)') return "\n".join(lines) if lines else "仓库无提交记录" @@ -1051,19 +1087,26 @@ def _parse_agent_action(raw): return {"action": "deliver", "result": raw} -def _resolve_repo_target(workspace_dir, repo_dir): - """解析仓库目录：空→repos/下第一个git仓库；'repos/xxx'→直接；'xxx'→repos/xxx。""" - repos_dir = os.path.join(workspace_dir, 'repos') +def _resolve_repo_target(space_dir, repo_dir): + """解析仓库目录：空→apps/modules 下第一个 git 仓库；'apps/xxx'/'modules/xxx'→直接；'xxx'→apps/xxx 或 modules/xxx。""" if repo_dir: - if repo_dir.startswith('repos/') or repo_dir.startswith('repos\\'): - return os.path.join(workspace_dir, repo_dir) - return os.path.join(repos_dir, repo_dir) - if os.path.isdir(repos_dir): - dirs = [d for d in os.listdir(repos_dir) - if os.path.isdir(os.path.join(repos_dir, d, '.git'))] - if dirs: - return os.path.join(repos_dir, dirs[0]) - return workspace_dir + if repo_dir.startswith('apps/') or repo_dir.startswith('apps\\') \ + or repo_dir.startswith('modules/') or repo_dir.startswith('modules\\'): + return os.path.join(space_dir, repo_dir) + # 无前缀：先试 apps/，再试 modules/ + for sub in ('apps', 'modules'): + cand = os.path.join(space_dir, sub, repo_dir) + if os.path.isdir(cand): + return cand + return os.path.join(space_dir, 'modules', repo_dir) + for sub in ('apps', 'modules'): + base = os.path.join(space_dir, sub) + if os.path.isdir(base): + dirs = [d for d in os.listdir(base) + if os.path.isdir(os.path.join(base, d, '.git'))] + if dirs: + return os.path.join(base, dirs[0]) + return space_dir def _agent_tools_to_openai_schema(agent_tools): @@ -1089,6 +1132,8 @@ def _agent_tools_to_openai_schema(agent_tools): async def _exec_agent_tool(tool, params, workspace_dir, ctx=None): + # 注意：workspace_dir 形参实际接收的是 space_dir（机构工作空间层 {space}/）， + # 角色相对路径 projects/{项目}/、apps/、modules/ 都以它为基准。 p = params or {} try: if tool == 'read_file': @@ -1137,7 +1182,10 @@ async def _exec_agent_tool(tool, params, workspace_dir, ctx=None): url = p.get('repo_url', '') if not url: return 'FAIL: 需要仓库URL' name = p.get('repo_name', '') or url.rstrip('/').split('/')[-1].replace('.git', '') - target = os.path.join(workspace_dir, 'repos', name) + if name.endswith('_app'): + target = os.path.join(workspace_dir, 'apps', name[:-4]) + else: + target = os.path.join(workspace_dir, 'modules', name) r = await _git_clone(url, target, p.get('branch', 'main')) return f"rc={r['rc']} {r['message']}" elif tool == 'git_status': @@ -1180,23 +1228,29 @@ _FORCE_PRODUCE_HINT = ( ) -async def _build_fallback_deliverable(workspace_dir, written_files, repos_dir, role): +async def _build_fallback_deliverable(space_dir, written_files, role): """循环结束未 deliver 时，检测 write_file/git 实际产出，构造真实交付件给 QC（而非占位符「Agent未产出交付件」）。""" lines = [] if written_files: lines.append("本任务执行期间 write_file 实际写入的文件：") for f in sorted(set(written_files)): - rel = os.path.relpath(f, workspace_dir) if f.startswith(workspace_dir) else f + rel = os.path.relpath(f, space_dir) if f.startswith(space_dir) else f lines.append("- " + rel) - if os.path.isdir(repos_dir): - for repo_name in sorted(os.listdir(repos_dir)): - repo_path = os.path.join(repos_dir, repo_name) - if os.path.isdir(os.path.join(repo_path, '.git')): + # 扫描 apps/ + modules/ + projects/ 下的 git 仓库工作区变更 + for sub in ('apps', 'modules', 'projects'): + base = os.path.join(space_dir, sub) + if not os.path.isdir(base): + continue + entries = ([(p, p) for p in sorted(os.listdir(base))] if sub == 'projects' + else [(f"{sub}/{n}", n) for n in sorted(os.listdir(base))]) + for label, name in entries: + rp = os.path.join(base, name) + if os.path.isdir(os.path.join(rp, '.git')): try: - r = await _run_shell('git status --short', repo_path, 10) + r = await _run_shell('git status --short', rp, 10) out = (r.get('stdout') or '').strip() if out: - lines.append(f"git 仓库 [{repo_name}] 工作区变更：") + lines.append(f"git 仓库 [{label}] 工作区变更：") lines.append(out[:2000]) except Exception: pass @@ -1361,12 +1415,9 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): except Exception: pass - repos_dir = os.path.join(workspace_dir, 'repos') - os.makedirs(repos_dir, exist_ok=True) - - # 确保项目关联仓库已 clone（幂等：已存在则 pull）——否则源码写不进 git 仓库 + # 确保项目关联仓库已 clone 到 apps/modules + 项目目录 git init（幂等）——否则源码写不进 git 仓库 try: - clone_results = await _setup_repos(sor, workspace_dir, project_id) + clone_results = await _setup_repos(sor, space_dir, project_id) logger.info(f"role_agent_run setup_repos: {clone_results}") except Exception as e: logger.warning(f"role_agent_run setup_repos failed: {e}") @@ -1382,6 +1433,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): role_skills = await _build_role_skills_block(sor, project_id, role, org_id) # 项目名（用于 prompt 里 {项目名}_pc 占位符替换，角色产出路径以 project-directory-spec 为准） project_name = await _get_project_name(sor, project_id) + project_dir = os.path.join(space_dir, 'projects', project_name) if project_name else space_dir # 能力工具上下文（project_id/iteration_id/who/agent_id 自动注入，LLM 不可见） _iter = None try: @@ -1515,7 +1567,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): if not deliverable: # 兜底：循环结束未 deliver，检测 write_file/git 实际产出构造真实交付件（而非占位符） - deliverable = await _build_fallback_deliverable(workspace_dir, written_files, repos_dir, role) + deliverable = await _build_fallback_deliverable(space_dir, written_files, role) # ── 处理产出 ── from appPublic.uniqueID import getID @@ -1546,7 +1598,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): # 写交付件文档 file_path = '' if result_text: - deliverable_dir = os.path.join(workspace_dir, 'deliverables', role) + deliverable_dir = os.path.join(project_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") @@ -1783,6 +1835,9 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): logger.warning(f"pm_review_run: org llm missing, task={task_id} org={org_id}") return {"status": "need_info", "task_id": task_id, "question_id": qid} workspace_dir = await _get_workspace_dir(sor, project_id) + space_dir = await _get_space_dir(sor, project_id) + project_name = await _get_project_name(sor, project_id) + project_dir = os.path.join(space_dir, 'projects', project_name) if project_name else space_dir try: await sor.sqlExe("COMMIT", {}) @@ -1791,7 +1846,7 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): repos = await _get_project_repos(sor, project_id) if repos: - await _setup_repos(sor, workspace_dir, project_id) + await _setup_repos(sor, space_dir, project_id) deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id) if not deliverable_content: @@ -1799,14 +1854,14 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): await reject_task(task_id, project_id, who="agent.pm", agent_id=agent_id, comment="没有交付件") return {"status": "rejected", "task_id": task_id, "reason": "没有交付件"} - repo_state = await _get_repo_state(workspace_dir) + repo_state = await _get_repo_state(space_dir, project_name) repos_str = ", ".join([r['name'] for r in repos]) if repos else "无" content_preview = deliverable_content[:6000] role_skills = await _build_role_skills_block(sor, project_id, _normalize_role('pm'), org_id) pm_system = PM_SYSTEM_PROMPT.replace('__TITLE__', title)\ .replace('__ROLE__', task_role)\ - .replace('__WORKSPACE__', workspace_dir)\ + .replace('__WORKSPACE__', space_dir)\ .replace('__REPOS__', repos_str)\ .replace('__REPO_STATE__', repo_state)\ .replace('__ROLE_SKILLS__', role_skills) @@ -1878,7 +1933,7 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): result = (f"已到最后收尾阶段（第 {turn + 1}/5 轮），拒绝执行探索类工具 {tool}。" f"请立即输出 review_approve / review_reject / review_complete / review_rollback 之一，不要再调用工具。") else: - result = await _exec_agent_tool(tool, params, workspace_dir) + result = await _exec_agent_tool(tool, params, space_dir) msgs.append({"role": "assistant", "content": raw}) msgs.append({"role": "user", "content": f"工具 {tool} 结果：\n{result}"}) else: @@ -1925,7 +1980,7 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): "id": pm_did, "project_id": project_id, "task_id": task_id, "deliverable_type": "pm_review", "title": f"PM审核：{title}", "content": json.dumps(decision, ensure_ascii=False), - "file_path": os.path.join(workspace_dir, 'deliverables', 'pm', f"{task_id}_review.md"), + "file_path": os.path.join(project_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}) @@ -1941,7 +1996,7 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): from .task_capability import approve_task await approve_task(task_id, project_id, who="agent.pm", agent_id=agent_id, comment=comment) # 审核通过后统一 git 提交（项目过程仓库 + 应用/模块仓库），不在每次生成时提交以减少并发锁 - _git_after = await _commit_repos_after_approve(workspace_dir, title) + _git_after = await _commit_repos_after_approve(space_dir, project_name, title) logger.info(f"commit-after-approve: task={task_id} {_git_after.get('message', '')}") next_role = await _get_next_role(task_role, project_id) if next_role: @@ -2008,7 +2063,7 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): "id": pm_did, "project_id": project_id, "task_id": task_id, "deliverable_type": "pm_review", "title": f"PM审核(回退)：{title}", "content": json.dumps(decision, ensure_ascii=False), - "file_path": os.path.join(workspace_dir, 'deliverables', 'pm', f"{task_id}_rollback.md"), + "file_path": os.path.join(project_dir, 'deliverables', 'pm', f"{task_id}_rollback.md"), "quality_score": 0, "review_status": "rejected", "created_by": agent_id or "pm", }) # 回退：作废回退点及之后的任务，创建回退目标的新任务 @@ -2063,6 +2118,7 @@ async def qc_review_run(project_id, agent_id=None, model_name=None): logger.warning(f"qc_review_run: org llm missing, task={task_id} org={org_id}") return {"status": "need_info", "task_id": task_id, "question_id": qid} workspace_dir = await _get_workspace_dir(sor, project_id) + space_dir = await _get_space_dir(sor, project_id) try: await sor.sqlExe("COMMIT", {}) @@ -2101,7 +2157,7 @@ async def qc_review_run(project_id, agent_id=None, model_name=None): qc_system = QC_SYSTEM_PROMPT.replace('__TITLE__', title)\ .replace('__ROLE__', task_role)\ - .replace('__WORKSPACE__', workspace_dir)\ + .replace('__WORKSPACE__', space_dir)\ .replace('__ROLE_SKILLS__', role_skills) msgs = [{"role": "system", "content": qc_system}] @@ -2140,7 +2196,7 @@ async def qc_review_run(project_id, agent_id=None, model_name=None): if tool == 'load_skill': result = await _load_skill_by_name(sor, project_id, 'qc', org_id, params.get('name', ''), params.get('file_path') or None) else: - result = await _exec_agent_tool(tool, params, workspace_dir) + result = await _exec_agent_tool(tool, params, space_dir) msgs.append({"role": "assistant", "content": raw}) msgs.append({"role": "user", "content": f"工具 {tool} 结果：\n{result}"}) else: diff --git a/pipeline_service/sdlc_ability.py b/pipeline_service/sdlc_ability.py index c901240..e534bdd 100644 --- a/pipeline_service/sdlc_ability.py +++ b/pipeline_service/sdlc_ability.py @@ -523,7 +523,9 @@ async def _h_approve_task(sor, p, ctx): if workspace_dir: try: from .agent_loop import _commit_repos_after_approve - await _commit_repos_after_approve(workspace_dir, title) + space_dir = os.path.dirname(workspace_dir.rstrip('/')) + project_name = os.path.basename(workspace_dir.rstrip('/')) + await _commit_repos_after_approve(space_dir, project_name, title) except Exception: pass @@ -829,18 +831,22 @@ async def _h_clone_repo(sor, p, ctx): if not pid: return "请先切换到项目" from pipeline_service.agent_loop import ( - _setup_repos, _get_workspace_dir, _git_clone) + _setup_repos, _get_workspace_dir, _get_space_dir, _git_clone) workspace_dir = await _get_workspace_dir(sor, pid) + space_dir = await _get_space_dir(sor, pid) url = (p.get("repo_url", "") or "").strip() if url: name = url.rstrip("/").split("/")[-1].replace(".git", "") - target = os.path.join(workspace_dir, "repos", name) + if name.endswith("_app"): + target = os.path.join(space_dir, "apps", name[:-4]) + else: + target = os.path.join(space_dir, "modules", name) r = await _git_clone(url, target, p.get("branch", "main")) return f"rc={r['rc']} {r['message']}" - results = await _setup_repos(sor, workspace_dir, pid) + results = await _setup_repos(sor, space_dir, pid) if not results: return "无关联仓库可克隆" return "\n".join(