yumoqing/pipeline-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(workspace): repos 逻辑迁到机构工作空间 projects/apps/modules——项目目录=项目过程仓库(git init)、应用/模块仓库按 _app 后缀分流 clone、commit/状态/解析扫 apps+modules+projects、交付件落 projects/{项目}/deliverables、PM/QC 工具路径基准改 space_dir

Browse Source
This commit is contained in:
ymq 2026-08-24 17:39:30 +08:00
parent b93c1796ca
commit ccf08bb631
2 changed files with 146 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

216
pipeline_service/agent_loop.py
View File

@ -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:

14
pipeline_service/sdlc_ability.py
View File

@ -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(