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

review修复：复盘认领校验补claimed_by比对、LLM硬超时兜底、报告落盘以write_file完整版为准

Browse Source
This commit is contained in:
ymq 2026-08-31 16:15:00 +08:00
parent 0547c37302
commit 65db2dddc3
1 changed files with 18 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
pipeline_service/agent_loop.py
View File

@ -3263,9 +3263,9 @@ async def retrospective_run(project_id, agent_id=None, model_name=None):
"UPDATE pipeline_tasks SET state='running', claimed_by=${cb}$, updated_at=NOW() " "UPDATE pipeline_tasks SET state='running', claimed_by=${cb}$, updated_at=NOW() "
"WHERE id=${tid}$ AND state='submitted' AND (claimed_by IS NULL OR claimed_by='')", "WHERE id=${tid}$ AND state='submitted' AND (claimed_by IS NULL OR claimed_by='')",
{"cb": claim_token, "tid": task_id}) {"cb": claim_token, "tid": task_id})
chk = await sor.sqlExe("SELECT state FROM pipeline_tasks WHERE id=${tid}$", {"tid": task_id}) chk = await sor.sqlExe("SELECT state, claimed_by FROM pipeline_tasks WHERE id=${tid}$", {"tid": task_id})
await sor.sqlExe("COMMIT", {}) await sor.sqlExe("COMMIT", {})
if not chk or getattr(chk[0], 'state', '') != 'running': if not chk or getattr(chk[0], 'state', '') != 'running' or getattr(chk[0], 'claimed_by', '') != claim_token:
return {"status": "idle", "message": "复盘任务认领竞争失败"} return {"status": "idle", "message": "复盘任务认领竞争失败"}
space_dir = await _get_space_dir(sor, project_id) space_dir = await _get_space_dir(sor, project_id)
@ -3304,7 +3304,9 @@ async def retrospective_run(project_id, agent_id=None, model_name=None):
msgs.append({"role": "user", "content": msgs.append({"role": "user", "content":
"已执行足够轮次。现在必须立即收尾：write_file 写复盘报告（若未写），然后输出 deliver 提交，禁止再调其它工具。"}) "已执行足够轮次。现在必须立即收尾：write_file 写复盘报告（若未写），然后输出 deliver 提交，禁止再调其它工具。"})
try: try:
raw = await llm_call_msgs(msgs, model=model_name, temperature=0.3, org_id=org_id) raw = await asyncio.wait_for(
llm_call_msgs(msgs, model=model_name, temperature=0.3, org_id=org_id),
timeout=_LLM_HARD_TIMEOUT)
except Exception as e: except Exception as e:
err_msg = f"{type(e).__name__}: {str(e)[:400]}" err_msg = f"{type(e).__name__}: {str(e)[:400]}"
from .task_capability import mark_failed from .task_capability import mark_failed
@ -3337,11 +3339,20 @@ async def retrospective_run(project_id, agent_id=None, model_name=None):
return {"status": "failed", "task_id": task_id, "error": "轮次耗尽未交付"} return {"status": "failed", "task_id": task_id, "error": "轮次耗尽未交付"}
# 复盘报告落盘：规范路径 + deliverables/pm/ # 复盘报告落盘：规范路径 + deliverables/pm/
# 报告优先取 LLM write_file 已写的完整版（工具轮写盘），deliver 摘要仅作兜底——
# 否则完整报告被摘要覆盖、五段结构丢失。
result_text = deliverable.get("result") or deliverable.get("summary") or "" result_text = deliverable.get("result") or deliverable.get("summary") or ""
report_rel = os.path.join('projects', project_name or '', 'docs', '02-retrospective', 'retrospective.md') report_rel = os.path.join('projects', project_name or '', 'docs', '02-retrospective', 'retrospective.md')
report_abs = os.path.join(space_dir, report_rel)
try: try:
os.makedirs(os.path.dirname(os.path.join(space_dir, report_rel)), exist_ok=True) if os.path.isfile(report_abs):
with open(os.path.join(space_dir, report_rel), 'w', encoding='utf-8') as f: with open(report_abs, encoding='utf-8') as f:
on_disk = f.read()
if len(on_disk.strip()) >= len(result_text.strip()):
result_text = on_disk # write_file 版更完整，以盘上为准
else:
os.makedirs(os.path.dirname(report_abs), exist_ok=True)
with open(report_abs, 'w', encoding='utf-8') as f:
f.write(result_text or '') f.write(result_text or '')
except Exception as e: except Exception as e:
logger.warning(f"retrospective report write failed: {e}") logger.warning(f"retrospective report write failed: {e}")
@ -3351,7 +3362,7 @@ async def retrospective_run(project_id, agent_id=None, model_name=None):
"id": did, "project_id": project_id, "task_id": task_id, "id": did, "project_id": project_id, "task_id": task_id,
"deliverable_type": "retrospective", "title": "项目复盘报告", "deliverable_type": "retrospective", "title": "项目复盘报告",
"content": result_text[:60000], "content": result_text[:60000],
"file_path": os.path.join(project_dir, 'deliverables', 'pm', f"{task_id}_retrospective.md"), "file_path": report_abs,
"quality_score": 100, "review_status": "approved", "created_by": agent_id or "pm", "quality_score": 100, "review_status": "approved", "created_by": agent_id or "pm",
}) })
await sor.sqlExe("COMMIT", {}) await sor.sqlExe("COMMIT", {})